diff --git a/framework/configstore/tables/webhooks.go b/framework/configstore/tables/webhooks.go index 98575053309..093a5961e1d 100644 --- a/framework/configstore/tables/webhooks.go +++ b/framework/configstore/tables/webhooks.go @@ -1,6 +1,7 @@ package tables import ( + "database/sql/driver" "encoding/json" "fmt" "net/url" @@ -37,6 +38,12 @@ func (e WebhookEvent) IsValid() bool { return slices.Contains(WebhookEvents, e) } +// Value implements driver.Valuer so database drivers that append typed column +// values (e.g. clickhouse-go batch inserts) can serialize the type. +func (e WebhookEvent) Value() (driver.Value, error) { + return string(e), nil +} + // TableWebhookEndpoint represents a registered webhook endpoint in the database. type TableWebhookEndpoint struct { ID string `gorm:"type:varchar(36);primaryKey" json:"id"` diff --git a/framework/logstore/logstoreparity_test.go b/framework/logstore/logstoreparity_test.go index 16d5cbbf342..4a8444be603 100644 --- a/framework/logstore/logstoreparity_test.go +++ b/framework/logstore/logstoreparity_test.go @@ -252,11 +252,13 @@ func canonicalizeOrder(key string, v any) any { tv[i] = canonicalizeOrder("", tv[i]) } if key == "rankings" { - keys := make([]string, len(tv)) - for i, e := range tv { - keys[i] = canonicalEntryKey(e) - } - sort.SliceStable(tv, func(i, j int) bool { return keys[i] < keys[j] }) + // Compute the key from the live element: a precomputed key slice + // would not be permuted alongside tv by SliceStable's swapper, so + // after the first swap the comparator would read stale keys and + // the final order would depend on the raw SQL row order. + sort.SliceStable(tv, func(i, j int) bool { + return canonicalEntryKey(tv[i]) < canonicalEntryKey(tv[j]) + }) } return tv default: diff --git a/framework/logstore/matview_count_test.go b/framework/logstore/matview_count_test.go index 4aae7a09e0e..467f111e50f 100644 --- a/framework/logstore/matview_count_test.go +++ b/framework/logstore/matview_count_test.go @@ -109,12 +109,12 @@ func TestSearchLogsMatViewCountCanonicalModelFilter(t *testing.T) { end := day.Add(35*time.Hour + 45*time.Minute) // 25h15m window, matview-eligible canonical := "gpt-4o-mini" - interior := day.Add(18 * time.Hour) // deep inside the window, never a boundary sliver - insertCountTestModelLog(t, db, interior, "success", canonical, nil) // wire-model match - insertCountTestModelLog(t, db, interior.Add(time.Minute), "success", "prod-alias", &canonical) // canonical-only match - insertCountTestModelLog(t, db, interior.Add(2*time.Minute), "success", "other-model", nil) // no match - insertCountTestModelLog(t, db, start.Add(5*time.Minute), "success", "prod-alias", &canonical) // canonical match in the head boundary sliver - insertCountTestModelLog(t, db, day.Add(9*time.Hour), "success", "prod-alias", &canonical) // before start -> excluded + interior := day.Add(18 * time.Hour) // deep inside the window, never a boundary sliver + insertCountTestModelLog(t, db, interior, "success", canonical, nil) // wire-model match + insertCountTestModelLog(t, db, interior.Add(time.Minute), "success", "prod-alias", &canonical) // canonical-only match + insertCountTestModelLog(t, db, interior.Add(2*time.Minute), "success", "other-model", nil) // no match + insertCountTestModelLog(t, db, start.Add(5*time.Minute), "success", "prod-alias", &canonical) // canonical match in the head boundary sliver + insertCountTestModelLog(t, db, day.Add(9*time.Hour), "success", "prod-alias", &canonical) // before start -> excluded refreshTestMatViews(t, db) store.matViewsReady.Store(true) @@ -253,3 +253,94 @@ func TestGetHistogramMatViewTrimsBoundaryBuckets(t *testing.T) { assert.Equal(t, int64(1), byBucket[day.Add(35*time.Hour).Unix()], "last bar holds only the in-window sliver row, not the post-end row from the same hour") } + +// insertCacheTestLog inserts a terminal log with an explicit cache_debug +// payload for the hybrid cache-hit tests. Pass nil for a row with no +// cache_debug at all. +func insertCacheTestLog(t *testing.T, db *gorm.DB, ts time.Time, cacheDebug *string) { + t.Helper() + err := db.Exec(` + INSERT INTO logs (id, timestamp, object_type, provider, model, status, cache_debug, + created_at, latency, cost, prompt_tokens, completion_tokens, total_tokens) + VALUES (?, ?, 'chat_completion', 'openai', 'gpt-4', 'success', ?, ?, 100, 0.01, 10, 5, 15) + `, uuid.New().String(), ts, cacheDebug, ts).Error + require.NoError(t, err, "failed to insert cache test log") +} + +// TestGetStatsMatViewCacheHitsHybrid verifies cache-hit stats come from the +// interior+boundary hybrid (materialized in mv_logs_hourly, classified raw in +// the slivers) instead of a full-window raw scan, and that the matview path +// agrees exactly with the raw path, including the nil contract: fields stay +// nil when no row in the window carried valid cache_debug JSON, and are +// explicit zeros when cache rows exist but none were direct/semantic. +func TestGetStatsMatViewCacheHitsHybrid(t *testing.T) { + store, db := setupPerfTestDB(t) + ctx := context.Background() + + day := time.Date(2026, 7, 20, 0, 0, 0, 0, time.UTC) + direct := `{"hit_type":"direct"}` + semantic := `{"hit_type":"semantic"}` + other := `{"hit_type":"external"}` + malformed := `not json` + + // Main window: 10:30 -> next day 11:45. + start := day.Add(10*time.Hour + 30*time.Minute) + end := day.Add(35*time.Hour + 45*time.Minute) + insertCacheTestLog(t, db, day.Add(10*time.Hour+15*time.Minute), &direct) // before start -> excluded + insertCacheTestLog(t, db, day.Add(10*time.Hour+45*time.Minute), &direct) // head sliver -> raw classifier + insertCacheTestLog(t, db, day.Add(18*time.Hour), &semantic) // interior -> matview column + insertCacheTestLog(t, db, day.Add(18*time.Hour+5*time.Minute), &other) // counts as cache row, neither type + insertCacheTestLog(t, db, day.Add(19*time.Hour), &malformed) // guard rejects -> not a cache row + insertCacheTestLog(t, db, day.Add(35*time.Hour+30*time.Minute), &direct) // tail sliver -> raw classifier + insertCacheTestLog(t, db, day.Add(35*time.Hour+50*time.Minute), &semantic) // after end -> excluded + + // Nil-contract window: 40h -> 70h holds one row with no cache_debug. + nilStart, nilEnd := day.Add(40*time.Hour), day.Add(70*time.Hour) + insertCacheTestLog(t, db, day.Add(50*time.Hour), nil) + + // Zero-contract window: 80h -> 110h holds only an other-hit_type cache row. + zeroStart, zeroEnd := day.Add(80*time.Hour), day.Add(110*time.Hour) + insertCacheTestLog(t, db, day.Add(90*time.Hour), &other) + + refreshTestMatViews(t, db) + store.matViewsReady.Store(true) + + filters := SearchFilters{StartTime: &start, EndTime: &end} + require.True(t, store.canUseMatViewForFreshAggregate(filters)) + stats, err := store.GetStats(ctx, filters) + require.NoError(t, err) + require.NotNil(t, stats.DirectCacheHits) + require.NotNil(t, stats.SemanticCacheHits) + assert.Equal(t, int64(2), *stats.DirectCacheHits, "head + tail sliver direct hits") + assert.Equal(t, int64(1), *stats.SemanticCacheHits, "interior semantic hit") + + // The raw path over the identical filters must produce identical values. + store.matViewsReady.Store(false) + rawStats, err := store.GetStats(ctx, filters) + require.NoError(t, err) + require.NotNil(t, rawStats.DirectCacheHits) + require.NotNil(t, rawStats.SemanticCacheHits) + assert.Equal(t, *rawStats.DirectCacheHits, *stats.DirectCacheHits) + assert.Equal(t, *rawStats.SemanticCacheHits, *stats.SemanticCacheHits) + store.matViewsReady.Store(true) + + // Nil contract: no valid cache_debug rows in the window -> fields omitted + // on both paths. + filters = SearchFilters{StartTime: &nilStart, EndTime: &nilEnd} + require.True(t, store.canUseMatViewForFreshAggregate(filters)) + stats, err = store.GetStats(ctx, filters) + require.NoError(t, err) + assert.Nil(t, stats.DirectCacheHits, "no cache rows -> nil, matching the raw path contract") + assert.Nil(t, stats.SemanticCacheHits) + + // Zero contract: cache rows exist but none direct/semantic -> explicit + // zeros on both paths. + filters = SearchFilters{StartTime: &zeroStart, EndTime: &zeroEnd} + require.True(t, store.canUseMatViewForFreshAggregate(filters)) + stats, err = store.GetStats(ctx, filters) + require.NoError(t, err) + require.NotNil(t, stats.DirectCacheHits, "cache rows present -> explicit zeros, not omission") + require.NotNil(t, stats.SemanticCacheHits) + assert.Equal(t, int64(0), *stats.DirectCacheHits) + assert.Equal(t, int64(0), *stats.SemanticCacheHits) +} diff --git a/framework/logstore/matviewheal.go b/framework/logstore/matviewheal.go new file mode 100644 index 00000000000..53791f0c445 --- /dev/null +++ b/framework/logstore/matviewheal.go @@ -0,0 +1,119 @@ +package logstore + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgconn" +) + +// This file implements runtime resilience for the materialized-view read +// path. Startup already repairs shape drift (repairMatViewShapes diffs the +// live catalog against matviewRequiredColumns and drops stale views), but a +// query can still hit a missing or stale-shaped view at runtime: a replica +// that lost the ensureMatViews advisory-lock race during a rolling deploy +// reads while the lock holder is mid-rebuild, or an operator drops a view. +// Rather than gating reads on catalog checks (the approach of 8354ca76, +// reverted in 9c3e4c7f3), the failing query itself signals staleness: the +// dispatch site serves that request from the raw logs table, the matview +// read path is disabled process-wide, and a single-flight background repair +// recreates and refreshes the views before re-enabling it. A premature +// ready=true from any source is therefore harmless - the next shape error +// restarts the cycle, and the system converges once shapes are current. + +// isMatViewShapeError reports whether err indicates a materialized view that +// is missing or has a stale shape, so the caller should fall back to the raw +// logs table. Deliberately excluded: 42883 (undefined_function - our readers +// use only built-ins, so that is a code bug we want loud) and 0A000 +// (cached-plan drift is structurally prevented by the two-pool connection +// lifecycle; masking it would hide a regression there). +func isMatViewShapeError(err error) bool { + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) { + return false + } + switch pgErr.Code { + case "42P01", // undefined_table: view dropped (mid-repair window, operator action) + "42703", // undefined_column: old-shape view still present (the #5384 failure) + "55000": // object_not_in_prerequisite_state: matview exists but is not populated + return true + } + return false +} + +// fallBackToRaw classifies err; on a matview shape error it disables the +// matview read path for subsequent requests, kicks the background self-heal, +// and returns true so the dispatch site falls through to its raw-table +// implementation for the current request. Any other error (including nil) +// returns false and the caller returns normally. +func (s *RDBLogStore) fallBackToRaw(err error) bool { + if err == nil || !isMatViewShapeError(err) { + return false + } + s.matViewsReady.Store(false) + if s.logger != nil { + s.logger.Warn(fmt.Sprintf("logstore: matview query failed with shape error, serving from raw tables until repaired: %s", err)) + } + s.triggerMatViewSelfHeal() + return true +} + +// matViewHealCooldown bounds how often a process attempts a background +// repair. No dedicated retry loop exists here because recovery after a failed +// heal is owned by the periodic refresher: startMatViewRefresher re-arms +// matViewsReady on its next successful tick, and it is guaranteed to be +// running whenever self-heal can trigger (a shape error requires the matview +// path to have been enabled, which requires the boot-time ensureMatViews that +// also starts the refresher). Future shape errors additionally re-trigger the +// heal, cooldown-limited - while broken, every request keeps succeeding via +// the raw fallback. +const matViewHealCooldown = 30 * time.Second + +// triggerMatViewSelfHeal starts a single-flight background repair: recreate +// any missing or stale matviews (ensureMatViews serializes cross-replica on +// the refresh advisory lock and handles shape diffing, drops, creates, and +// index builds), refresh them, and re-enable the matview read path. Repair +// failures are logged only; the raw fallback keeps serving in the meantime. +// +// A replica that loses the advisory-lock race re-enables the read path while +// the lock holder may still be mid-rebuild; that premature enable is accepted +// - the next query against a still-stale view falls back raw and re-triggers +// the heal, converging once the lock holder finishes. +func (s *RDBLogStore) triggerMatViewSelfHeal() { + if !s.matViewHealInFlight.CompareAndSwap(false, true) { + return // a heal is already running + } + go func() { + defer s.matViewHealInFlight.Store(false) + if time.Since(time.Unix(0, s.matViewHealLastAttempt.Load())) < matViewHealCooldown { + return + } + s.matViewHealLastAttempt.Store(time.Now().UnixNano()) + + ctx := context.Background() + if err := ensureMatViews(ctx, s.db); err != nil { + if s.logger != nil { + s.logger.Warn(fmt.Sprintf("logstore: matview self-heal creation failed: %s (still serving from raw tables)", err)) + } + return + } + if err := refreshMatViews(ctx, s.db); err != nil { + if s.logger != nil { + s.logger.Warn(fmt.Sprintf("logstore: matview self-heal refresh failed: %s (still serving from raw tables)", err)) + } + return + } + s.matViewsReady.Store(true) + if s.logger != nil { + s.logger.Info("logstore: materialized views self-healed after shape error") + } + }() +} + +// resetMatViewHeal clears the single-flight and cooldown state. Test helper. +func (s *RDBLogStore) resetMatViewHeal() { + s.matViewHealInFlight.Store(false) + s.matViewHealLastAttempt.Store(0) +} diff --git a/framework/logstore/matviewheal_test.go b/framework/logstore/matviewheal_test.go new file mode 100644 index 00000000000..b4355790282 --- /dev/null +++ b/framework/logstore/matviewheal_test.go @@ -0,0 +1,160 @@ +package logstore + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/gorm" +) + +func TestIsMatViewShapeError(t *testing.T) { + shape := []error{ + &pgconn.PgError{Code: "42P01"}, // undefined_table + &pgconn.PgError{Code: "42703"}, // undefined_column + &pgconn.PgError{Code: "55000"}, // object_not_in_prerequisite_state + fmt.Errorf("query failed: %w", &pgconn.PgError{Code: "42703"}), // wrapped + } + for _, err := range shape { + assert.True(t, isMatViewShapeError(err), "expected shape error: %v", err) + } + + notShape := []error{ + nil, + context.Canceled, + gorm.ErrRecordNotFound, + errors.New("column does not exist"), // text without a PgError is not classified + &pgconn.PgError{Code: "0A000"}, // cached-plan drift must stay loud + &pgconn.PgError{Code: "42883"}, // undefined_function is a code bug + fmt.Errorf("wrap: %w", errors.New("42703")), // code in message only + } + for _, err := range notShape { + assert.False(t, isMatViewShapeError(err), "expected non-shape error: %v", err) + } +} + +// TestMatViewShapeErrorFallsBackAndSelfHeals drops mv_logs_hourly mid-run and +// verifies that (a) matview-gated reads keep returning correct results from +// the raw table with no error, (b) the matview read path is disabled +// immediately, and (c) the background self-heal recreates the view and +// re-enables the matview path without a restart. +func TestMatViewShapeErrorFallsBackAndSelfHeals(t *testing.T) { + store, db := setupPerfTestDB(t) + ctx := context.Background() + + day := time.Date(2026, 7, 20, 0, 0, 0, 0, time.UTC) + insertCountTestLog(t, db, day.Add(10*time.Hour), "success") + insertCountTestLog(t, db, day.Add(11*time.Hour), "error") + refreshTestMatViews(t, db) + store.matViewsReady.Store(true) + store.resetMatViewHeal() + + require.NoError(t, db.Exec("DROP MATERIALIZED VIEW mv_logs_hourly CASCADE").Error) + + // Unbounded window keeps the fresh-aggregate gate satisfied, so both + // calls attempt the matview first and must fall through on 42P01. + stats, err := store.GetStats(ctx, SearchFilters{}) + require.NoError(t, err, "GetStats must serve from raw tables when the view is missing") + assert.Equal(t, int64(2), stats.TotalRequests) + + result, err := store.SearchLogs(ctx, SearchFilters{}, PaginationOptions{Limit: 10}) + require.NoError(t, err, "SearchLogs must serve from raw tables when the view is missing") + assert.Equal(t, int64(2), result.Pagination.TotalCount) + + assert.False(t, store.matViewsReady.Load(), + "the matview read path must be disabled after a shape error") + + require.Eventually(t, func() bool { + if !store.matViewsReady.Load() { + return false + } + var exists bool + if err := db.Raw( + "SELECT EXISTS (SELECT 1 FROM pg_class WHERE relname = 'mv_logs_hourly' AND relkind = 'm')", + ).Scan(&exists).Error; err != nil { + return false + } + return exists + }, 90*time.Second, 500*time.Millisecond, + "self-heal must recreate mv_logs_hourly and re-enable the matview path") + + stats, err = store.GetStats(ctx, SearchFilters{}) + require.NoError(t, err) + assert.Equal(t, int64(2), stats.TotalRequests, "healed matview path must agree with raw") +} + +// TestMatViewStaleShapeFallsBackAndSelfHeals replaces mv_logs_hourly with an +// old-shape view missing required columns (the #5384 rolling-deploy state: +// REFRESH succeeds on it, only reads fail) and verifies the 42703 path: raw +// fallback with no error, then background repair via repairMatViewShapes. +func TestMatViewStaleShapeFallsBackAndSelfHeals(t *testing.T) { + store, db := setupPerfTestDB(t) + ctx := context.Background() + + day := time.Date(2026, 7, 20, 0, 0, 0, 0, time.UTC) + insertCountTestLog(t, db, day.Add(10*time.Hour), "success") + + require.NoError(t, db.Exec("DROP MATERIALIZED VIEW mv_logs_hourly CASCADE").Error) + require.NoError(t, db.Exec(` + CREATE MATERIALIZED VIEW mv_logs_hourly AS + SELECT date_trunc('hour', timestamp) AS hour, provider, model, status, COUNT(*) AS count + FROM logs WHERE status IN ('success', 'error', 'cancelled') + GROUP BY 1, 2, 3, 4 + `).Error) + // REFRESH succeeds on the stale shape - readiness alone cannot detect it. + require.NoError(t, db.Exec("REFRESH MATERIALIZED VIEW mv_logs_hourly").Error) + store.matViewsReady.Store(true) + store.resetMatViewHeal() + + hist, err := store.GetHistogram(ctx, SearchFilters{}, 3600) + require.NoError(t, err, "GetHistogram must serve from raw tables when the view shape is stale") + var total int64 + for _, b := range hist.Buckets { + total += b.Count + } + assert.Equal(t, int64(1), total) + assert.False(t, store.matViewsReady.Load()) + + require.Eventually(t, func() bool { + if !store.matViewsReady.Load() { + return false + } + var hasColumn bool + if err := db.Raw(` + SELECT EXISTS ( + SELECT 1 FROM pg_attribute a + JOIN pg_class c ON c.oid = a.attrelid + WHERE c.relname = 'mv_logs_hourly' AND c.relkind = 'm' + AND a.attname = 'cancelled_count' AND a.attnum > 0 AND NOT a.attisdropped + )`).Scan(&hasColumn).Error; err != nil { + return false + } + return hasColumn + }, 90*time.Second, 500*time.Millisecond, + "self-heal must rebuild the stale-shaped view with the current columns") +} + +// TestFilterMatViewShapeErrorFallsBack covers the mv_filter_* views: dropping +// one must not break its GetDistinct* endpoint. +func TestFilterMatViewShapeErrorFallsBack(t *testing.T) { + store, db := setupPerfTestDB(t) + ctx := context.Background() + + day := time.Date(2026, 7, 20, 0, 0, 0, 0, time.UTC) + insertCountTestLog(t, db, day.Add(10*time.Hour), "success") // model gpt-4 + refreshTestMatViews(t, db) + store.matViewsReady.Store(true) + store.resetMatViewHeal() + + require.NoError(t, db.Exec("DROP MATERIALIZED VIEW mv_filter_models CASCADE").Error) + + models, err := store.GetDistinctModels(ctx, 10, "") + require.NoError(t, err, "GetDistinctModels must serve from raw tables when the filter view is missing") + assert.Contains(t, models, "gpt-4") + assert.False(t, store.matViewsReady.Load()) +} diff --git a/framework/logstore/matviews.go b/framework/logstore/matviews.go index 6a25676f0eb..124a094f73a 100644 --- a/framework/logstore/matviews.go +++ b/framework/logstore/matviews.go @@ -62,12 +62,34 @@ SELECT COALESCE(COUNT(*) FILTER (WHERE latency > 0), 0) AS throughput_request_count, COALESCE(SUM(total_tokens), 0) AS total_tokens, COALESCE(SUM(cached_read_tokens), 0) AS total_cached_read_tokens, - COALESCE(SUM(cost), 0) AS total_cost + COALESCE(SUM(cost), 0) AS total_cost, + -- Cache-hit measures precomputed from cache_debug so /api/logs/stats can + -- serve them from the hybrid instead of a full-window raw scan. Safe to + -- materialize: cache_debug is written with the terminal status and never + -- mutated afterwards. cache_debug_count exists to preserve the raw path's + -- nil contract (see getStatsFromMatView): it distinguishes "no rows had + -- valid cache_debug at all" (fields omitted) from "cache rows exist but + -- none were direct/semantic" (explicit zeros). + SUM(CASE WHEN ` + cacheDebugJSONGuard + ` + AND ` + cacheDebugHitTypeExpr + ` = 'direct' THEN 1 ELSE 0 END) AS direct_cache_hits, + SUM(CASE WHEN ` + cacheDebugJSONGuard + ` + AND ` + cacheDebugHitTypeExpr + ` = 'semantic' THEN 1 ELSE 0 END) AS semantic_cache_hits, + COUNT(*) FILTER (WHERE ` + cacheDebugJSONGuard + `) AS cache_debug_count FROM logs WHERE status IN ('success', 'error', 'cancelled') GROUP BY 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14 ` +// cacheDebugJSONGuard matches rows whose cache_debug column holds a loose +// JSON object. Shared by the matview DDL, the hybrid boundary aggregate, and +// aggregateCacheHits' postgres branch so every classifier selects the same +// population. +const cacheDebugJSONGuard = `cache_debug IS NOT NULL AND cache_debug <> '' AND cache_debug ~ '^\s*\{.*\}\s*$'` + +// cacheDebugHitTypeExpr extracts the hit_type value from the cache_debug +// JSON. POSIX regex only, so it is valid inside a matview definition. +const cacheDebugHitTypeExpr = `substring(cache_debug from '"hit_type"[[:space:]]*:[[:space:]]*"([^"]+)"')` + // mvLogsHourlyUniqueIdx is required for REFRESH MATERIALIZED VIEW CONCURRENTLY. // CONCURRENTLY avoids the AccessExclusiveLock that the plain form would take // during startup ensure / repair paths. @@ -98,6 +120,9 @@ var mvLogsHourlyRequiredColumns = []string{ "throughput_completion_tokens", "throughput_latency_ms", "throughput_request_count", + "direct_cache_hits", + "semantic_cache_hits", + "cache_debug_count", } // legacyMatViewNames are matviews from previous schema versions that no longer @@ -1083,13 +1108,16 @@ func (s *RDBLogStore) countRawNonTerminal(ctx context.Context, filters SearchFil // the total latency over counted rows (SUM(avg_latency*count) on the matview, // SUM(latency) on raw), divided once at the end for the combined average. type matViewStatsAgg struct { - Count int64 `gorm:"column:total_count"` - SuccessCount int64 `gorm:"column:success_count"` - LatencySum float64 `gorm:"column:latency_sum"` - TotalTokens int64 `gorm:"column:total_tokens"` - PromptTokens int64 `gorm:"column:prompt_tokens"` - CompletionTokens int64 `gorm:"column:completion_tokens"` - TotalCost float64 `gorm:"column:total_cost"` + Count int64 `gorm:"column:total_count"` + SuccessCount int64 `gorm:"column:success_count"` + LatencySum float64 `gorm:"column:latency_sum"` + TotalTokens int64 `gorm:"column:total_tokens"` + PromptTokens int64 `gorm:"column:prompt_tokens"` + CompletionTokens int64 `gorm:"column:completion_tokens"` + TotalCost float64 `gorm:"column:total_cost"` + DirectCacheHits int64 `gorm:"column:direct_cache_hits"` + SemanticCacheHits int64 `gorm:"column:semantic_cache_hits"` + CacheDebugCount int64 `gorm:"column:cache_debug_count"` } func (a *matViewStatsAgg) add(b matViewStatsAgg) { @@ -1100,6 +1128,9 @@ func (a *matViewStatsAgg) add(b matViewStatsAgg) { a.PromptTokens += b.PromptTokens a.CompletionTokens += b.CompletionTokens a.TotalCost += b.TotalCost + a.DirectCacheHits += b.DirectCacheHits + a.SemanticCacheHits += b.SemanticCacheHits + a.CacheDebugCount += b.CacheDebugCount } // matViewInteriorStatsAgg sums the additive stat fields over the hour buckets @@ -1116,7 +1147,10 @@ func (s *RDBLogStore) matViewInteriorStatsAgg(ctx context.Context, dimFilters Se COALESCE(SUM(total_tokens), 0) AS total_tokens, COALESCE(SUM(total_prompt_tokens), 0) AS prompt_tokens, COALESCE(SUM(total_completion_tokens), 0) AS completion_tokens, - COALESCE(SUM(total_cost), 0) AS total_cost + COALESCE(SUM(total_cost), 0) AS total_cost, + COALESCE(SUM(direct_cache_hits), 0) AS direct_cache_hits, + COALESCE(SUM(semantic_cache_hits), 0) AS semantic_cache_hits, + COALESCE(SUM(cache_debug_count), 0) AS cache_debug_count `).Scan(&agg).Error return agg, err } @@ -1140,7 +1174,10 @@ func (s *RDBLogStore) rawTerminalStatsAgg(ctx context.Context, dimFilters Search COALESCE(SUM(total_tokens), 0) AS total_tokens, COALESCE(SUM(prompt_tokens), 0) AS prompt_tokens, COALESCE(SUM(completion_tokens), 0) AS completion_tokens, - COALESCE(SUM(cost), 0) AS total_cost + COALESCE(SUM(cost), 0) AS total_cost, + COALESCE(SUM(CASE WHEN ` + cacheDebugJSONGuard + ` AND ` + cacheDebugHitTypeExpr + ` = 'direct' THEN 1 ELSE 0 END), 0) AS direct_cache_hits, + COALESCE(SUM(CASE WHEN ` + cacheDebugJSONGuard + ` AND ` + cacheDebugHitTypeExpr + ` = 'semantic' THEN 1 ELSE 0 END), 0) AS semantic_cache_hits, + COUNT(*) FILTER (WHERE ` + cacheDebugJSONGuard + `) AS cache_debug_count `).Scan(&agg).Error return agg, err } @@ -1217,18 +1254,16 @@ func (s *RDBLogStore) getStatsFromMatView(ctx context.Context, filters SearchFil CacheHitRateTotalRequests: &completed, } - // cache_debug is not stored in the matview — query the raw logs table - // directly over the exact window. The denominator (the hybrid terminal - // count) is exact-range now, so no hour alignment is needed; this also - // avoids Go-side time.Truncate, which would misalign with the SQL - // date_trunc grid on servers with fractional-hour timezone offsets. - cacheBase := s.ScopedDB(ctx).Model(&Log{}).Where("status IN ?", terminalLogStatuses) - direct, semantic, err := s.aggregateCacheHits(ctx, cacheBase, filters) - if err != nil { - s.logger.Warn(fmt.Sprintf("logstore: failed to aggregate cache-hit stats, skipping: %s", err)) - } else if direct != nil || semantic != nil { - stats.DirectCacheHits = direct - stats.SemanticCacheHits = semantic + // Cache hits come from the same hybrid aggregate (materialized in + // mv_logs_hourly for interior buckets, classified raw for the slivers) - + // no more full-window raw scan. CacheDebugCount reproduces + // aggregateCacheHits' nil contract: when no row in the window carried + // valid cache_debug JSON, the fields stay nil so they are omitted from + // the JSON payload, matching the raw path. + if agg.CacheDebugCount > 0 { + direct, semantic := agg.DirectCacheHits, agg.SemanticCacheHits + stats.DirectCacheHits = &direct + stats.SemanticCacheHits = &semantic } return stats, nil @@ -2102,7 +2137,7 @@ func (s *RDBLogStore) getModelRankingsFromMatView(ctx context.Context, filters S COALESCE(SUM(CASE WHEN status = 'success' THEN throughput_completion_tokens ELSE 0 END), 0) AS tp_completion_tokens, COALESCE(SUM(CASE WHEN status = 'success' THEN throughput_latency_ms ELSE 0 END), 0) AS tp_latency_ms `).Group("model, provider"). - Order("total DESC"). + Order("total DESC, model ASC, provider ASC"). Find(&results).Error; err != nil { return nil, err } @@ -2214,7 +2249,7 @@ func (s *RDBLogStore) getUserRankingsFromMatView(ctx context.Context, filters Se SUM(total_tokens) AS total_tkns, SUM(total_cost) AS total_cost `).Group("user_id"). - Order("total DESC"). + Order("total DESC, user_id ASC"). Find(&results).Error; err != nil { return nil, err } @@ -2310,7 +2345,7 @@ func (s *RDBLogStore) getDimensionRankingsFromMatView(ctx context.Context, filte SUM(total_tokens) AS total_tkns, SUM(total_cost) AS total_cost `, idCol)).Group(idCol). - Order("total DESC"). + Order(fmt.Sprintf("total DESC, %s ASC", idCol)). Find(&results).Error; err != nil { return nil, err } diff --git a/framework/logstore/rdb.go b/framework/logstore/rdb.go index 424cd17e17c..d0f258c3e03 100644 --- a/framework/logstore/rdb.go +++ b/framework/logstore/rdb.go @@ -57,6 +57,9 @@ type RDBLogStore struct { db *gorm.DB logger schemas.Logger matViewsReady atomic.Bool + // Self-heal state for the matview read path (see matviewheal.go). + matViewHealInFlight atomic.Bool + matViewHealLastAttempt atomic.Int64 // unix nanos of the last repair attempt } // generateBucketTimestamps generates all bucket timestamps for a time range. @@ -342,7 +345,7 @@ func (s *RDBLogStore) applyFilters(baseQuery *gorm.DB, filters SearchFilters) *g case "postgres": // Match the same loose-JSON guard used by aggregateCacheHits so the regex extract is safe. baseQuery = baseQuery.Where( - "cache_debug IS NOT NULL AND cache_debug <> '' AND cache_debug ~ '^\\s*\\{.*\\}\\s*$' AND substring(cache_debug from '\"hit_type\"[[:space:]]*:[[:space:]]*\"([^\"]+)\"') IN ?", + cacheDebugJSONGuard+" AND "+cacheDebugHitTypeExpr+" IN ?", valid, ) case "clickhouse": @@ -725,9 +728,11 @@ func (s *RDBLogStore) SearchLogs(ctx context.Context, filters SearchFilters, pag // keep the matview win because raw COUNT over multi-day ranges is the // expensive path. if s.db.Dialector.Name() == "postgres" && s.canUseMatViewForFreshAggregate(filters) { - var err error - totalCount, err = s.getCountFromMatView(gCtx, filters) - return err + c, err := s.getCountFromMatView(gCtx, filters) + if !s.fallBackToRaw(err) { + totalCount = c + return err + } } countQuery := s.ScopedDB(gCtx).Model(&Log{}) countQuery = s.applyFilters(countQuery, filters) @@ -1011,7 +1016,9 @@ func (s *RDBLogStore) GetStats(ctx context.Context, filters SearchFilters) (*Sea // consistent with the real-time /api/logs row list. See // canUseMatViewForFreshAggregate. if s.db.Dialector.Name() == "postgres" && s.canUseMatViewForFreshAggregate(filters) { - return s.getStatsFromMatView(ctx, filters) + if stats, err := s.getStatsFromMatView(ctx, filters); !s.fallBackToRaw(err) { + return stats, err + } } baseQuery := s.ScopedDB(ctx).Model(&Log{}) baseQuery = s.applyFilters(baseQuery, filters) @@ -1152,10 +1159,10 @@ func (s *RDBLogStore) aggregateCacheHits(ctx context.Context, base *gorm.DB, fil q := s.applyFilters(base, filters) switch s.db.Dialector.Name() { case "postgres": - q = q.Where("cache_debug IS NOT NULL AND cache_debug <> '' AND cache_debug ~ '^\\s*\\{.*\\}\\s*$'") + q = q.Where(cacheDebugJSONGuard) if err := q.Select( - `SUM(CASE WHEN substring(cache_debug from '"hit_type"[[:space:]]*:[[:space:]]*"([^"]+)"') = 'direct' THEN 1 ELSE 0 END) AS direct_hits, ` + - `SUM(CASE WHEN substring(cache_debug from '"hit_type"[[:space:]]*:[[:space:]]*"([^"]+)"') = 'semantic' THEN 1 ELSE 0 END) AS semantic_hits`, + `SUM(CASE WHEN ` + cacheDebugHitTypeExpr + ` = 'direct' THEN 1 ELSE 0 END) AS direct_hits, ` + + `SUM(CASE WHEN ` + cacheDebugHitTypeExpr + ` = 'semantic' THEN 1 ELSE 0 END) AS semantic_hits`, ).Scan(&result).Error; err != nil { return nil, nil, fmt.Errorf("failed to aggregate cache-hit stats: %w", err) } @@ -1190,7 +1197,9 @@ func (s *RDBLogStore) GetHistogram(ctx context.Context, filters SearchFilters, b bucketSizeSeconds = 3600 // Default to 1 hour } if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { - return s.getHistogramFromMatView(ctx, filters, bucketSizeSeconds) + if res, err := s.getHistogramFromMatView(ctx, filters, bucketSizeSeconds); !s.fallBackToRaw(err) { + return res, err + } } // Determine database type for SQL syntax @@ -1301,8 +1310,15 @@ func (s *RDBLogStore) GetTokenHistogram(ctx context.Context, filters SearchFilte if bucketSizeSeconds <= 0 { bucketSizeSeconds = 3600 // Default to 1 hour } +<<<<<<< HEAD if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getTokenHistogramFromMatView(ctx, filters, bucketSizeSeconds) +======= + if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getTokenHistogramFromMatView(ctx, filters, bucketSizeSeconds); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } dialect := s.db.Dialector.Name() @@ -1420,8 +1436,15 @@ func (s *RDBLogStore) GetThroughputHistogram(ctx context.Context, filters Search if bucketSizeSeconds <= 0 { bucketSizeSeconds = 3600 // Default to 1 hour } +<<<<<<< HEAD if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getThroughputHistogramFromMatView(ctx, filters, bucketSizeSeconds) +======= + if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getThroughputHistogramFromMatView(ctx, filters, bucketSizeSeconds); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } dialect := s.db.Dialector.Name() @@ -1507,8 +1530,15 @@ func (s *RDBLogStore) GetProviderThroughputHistogram(ctx context.Context, filter if bucketSizeSeconds <= 0 { bucketSizeSeconds = 3600 } +<<<<<<< HEAD if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getProviderThroughputHistogramFromMatView(ctx, filters, bucketSizeSeconds) +======= + if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getProviderThroughputHistogramFromMatView(ctx, filters, bucketSizeSeconds); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } dialect := s.db.Dialector.Name() @@ -1600,8 +1630,15 @@ func (s *RDBLogStore) GetCostHistogram(ctx context.Context, filters SearchFilter if bucketSizeSeconds <= 0 { bucketSizeSeconds = 3600 // Default to 1 hour } +<<<<<<< HEAD if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getCostHistogramFromMatView(ctx, filters, bucketSizeSeconds) +======= + if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getCostHistogramFromMatView(ctx, filters, bucketSizeSeconds); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } dialect := s.db.Dialector.Name() @@ -1706,8 +1743,15 @@ func (s *RDBLogStore) GetModelHistogram(ctx context.Context, filters SearchFilte if bucketSizeSeconds <= 0 { bucketSizeSeconds = 3600 // Default to 1 hour } +<<<<<<< HEAD if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getModelHistogramFromMatView(ctx, filters, bucketSizeSeconds) +======= + if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getModelHistogramFromMatView(ctx, filters, bucketSizeSeconds); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } dialect := s.db.Dialector.Name() @@ -1845,8 +1889,15 @@ func (s *RDBLogStore) GetLatencyHistogram(ctx context.Context, filters SearchFil if bucketSizeSeconds <= 0 { bucketSizeSeconds = 3600 } +<<<<<<< HEAD if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getLatencyHistogramFromMatView(ctx, filters, bucketSizeSeconds) +======= + if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getLatencyHistogramFromMatView(ctx, filters, bucketSizeSeconds); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } dialect := s.db.Dialector.Name() @@ -2110,7 +2161,9 @@ func (s *RDBLogStore) buildLatencyHistogramResult(computedBuckets map[int64]Late // cost-histogram totals shown on the same dashboard. func (s *RDBLogStore) GetModelRankings(ctx context.Context, filters SearchFilters) (*ModelRankingResult, error) { if s.db.Dialector.Name() == "postgres" && s.canUseMatViewForFreshAggregate(filters) { - return s.getModelRankingsFromMatView(ctx, filters) + if res, err := s.getModelRankingsFromMatView(ctx, filters); !s.fallBackToRaw(err) { + return res, err + } } selectClause := ` model, @@ -2149,7 +2202,7 @@ func (s *RDBLogStore) GetModelRankings(ctx context.Context, filters SearchFilter if err := currentQuery. Select(currentSelectClause). Group("model, provider"). - Order("total_requests DESC"). + Order("total_requests DESC, model ASC, provider ASC"). Limit(defaultMaxRankingsLimit). Find(¤tResults).Error; err != nil { return nil, fmt.Errorf("failed to get model rankings: %w", err) @@ -2272,7 +2325,9 @@ func (s *RDBLogStore) GetModelRankings(ctx context.Context, filters SearchFilter // cost-histogram totals shown on the same dashboard. func (s *RDBLogStore) GetUserRankings(ctx context.Context, filters SearchFilters) (*UserRankingResult, error) { if s.db.Dialector.Name() == "postgres" && s.canUseMatViewForFreshAggregate(filters) { - return s.getUserRankingsFromMatView(ctx, filters) + if res, err := s.getUserRankingsFromMatView(ctx, filters); !s.fallBackToRaw(err) { + return res, err + } } selectClause := ` user_id, @@ -2297,7 +2352,7 @@ func (s *RDBLogStore) GetUserRankings(ctx context.Context, filters SearchFilters if err := currentQuery. Select(selectClause). Group("user_id"). - Order("total_requests DESC"). + Order("total_requests DESC, user_id ASC"). Limit(defaultMaxRankingsLimit). Find(¤tResults).Error; err != nil { return nil, fmt.Errorf("failed to get user rankings: %w", err) @@ -2402,7 +2457,9 @@ func (s *RDBLogStore) GetDimensionRankings(ctx context.Context, filters SearchFi // cost-histogram totals shown on the same dashboard. Bucketed dimensions // always use the raw path — the matview reader has no Unassigned bucket. if !bucketed && s.db.Dialector.Name() == "postgres" && s.canUseMatViewForFreshAggregate(filters) { - return s.getDimensionRankingsFromMatView(ctx, filters, dimension) + if res, err := s.getDimensionRankingsFromMatView(ctx, filters, dimension); !s.fallBackToRaw(err) { + return res, err + } } var nameExpr string @@ -2438,7 +2495,11 @@ func (s *RDBLogStore) GetDimensionRankings(ctx context.Context, filters SearchFi if err := currentQuery. Select(selectClause). Group(groupExpr). - Order("total_requests DESC"). + // Tiebreak on the group expression itself, not the `id` alias: + // ClickHouse resolves a bare `id` in ORDER BY to the base table's + // column (not in GROUP BY -> error 215), while Postgres/SQLite + // resolve the alias. The expression works on all three. + Order("total_requests DESC, " + groupExpr + " ASC"). Limit(defaultMaxRankingsLimit). Find(¤tResults).Error; err != nil { return nil, fmt.Errorf("failed to get dimension rankings for %s: %w", dimension, err) @@ -2577,8 +2638,15 @@ func (s *RDBLogStore) GetProviderCostHistogram(ctx context.Context, filters Sear if bucketSizeSeconds <= 0 { bucketSizeSeconds = 3600 } +<<<<<<< HEAD if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getProviderCostHistogramFromMatView(ctx, filters, bucketSizeSeconds) +======= + if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getProviderCostHistogramFromMatView(ctx, filters, bucketSizeSeconds); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } dialect := s.db.Dialector.Name() @@ -2672,8 +2740,15 @@ func (s *RDBLogStore) GetProviderTokenHistogram(ctx context.Context, filters Sea if bucketSizeSeconds <= 0 { bucketSizeSeconds = 3600 } +<<<<<<< HEAD if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getProviderTokenHistogramFromMatView(ctx, filters, bucketSizeSeconds) +======= + if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getProviderTokenHistogramFromMatView(ctx, filters, bucketSizeSeconds); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } dialect := s.db.Dialector.Name() @@ -2779,8 +2854,15 @@ func (s *RDBLogStore) GetProviderLatencyHistogram(ctx context.Context, filters S if bucketSizeSeconds <= 0 { bucketSizeSeconds = 3600 } +<<<<<<< HEAD if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getProviderLatencyHistogramFromMatView(ctx, filters, bucketSizeSeconds) +======= + if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getProviderLatencyHistogramFromMatView(ctx, filters, bucketSizeSeconds); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } dialect := s.db.Dialector.Name() @@ -3143,8 +3225,15 @@ func (s *RDBLogStore) GetDimensionCostHistogram(ctx context.Context, filters Sea dimValueExpr = bucketedIDExpr(dimCol) groupCol = dimValueExpr } +<<<<<<< HEAD if !bucketed && dialect == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getDimensionCostHistogramFromMatView(ctx, filters, bucketSizeSeconds, dimension) +======= + if !bucketed && dialect == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getDimensionCostHistogramFromMatView(ctx, filters, bucketSizeSeconds, dimension); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } baseQuery := s.ScopedDB(ctx).Model(&Log{}) baseQuery = s.applyFilters(baseQuery, filters) @@ -3243,8 +3332,15 @@ func (s *RDBLogStore) GetDimensionTokenHistogram(ctx context.Context, filters Se dimValueExpr = bucketedIDExpr(dimCol) groupCol = dimValueExpr } +<<<<<<< HEAD if !bucketed && dialect == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getDimensionTokenHistogramFromMatView(ctx, filters, bucketSizeSeconds, dimension) +======= + if !bucketed && dialect == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getDimensionTokenHistogramFromMatView(ctx, filters, bucketSizeSeconds, dimension); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } baseQuery := s.ScopedDB(ctx).Model(&Log{}) baseQuery = s.applyFilters(baseQuery, filters) @@ -3352,8 +3448,15 @@ func (s *RDBLogStore) GetDimensionLatencyHistogram(ctx context.Context, filters if bucketSizeSeconds <= 0 { bucketSizeSeconds = 3600 } +<<<<<<< HEAD if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 && bucketSizeSeconds%3600 == 0 { return s.getDimensionLatencyHistogramFromMatView(ctx, filters, bucketSizeSeconds, dimension) +======= + if s.db.Dialector.Name() == "postgres" && s.canUseMatView(filters) && bucketSizeSeconds >= 3600 { + if res, err := s.getDimensionLatencyHistogramFromMatView(ctx, filters, bucketSizeSeconds, dimension); !s.fallBackToRaw(err) { + return res, err + } +>>>>>>> 5b4b7fe74 (add cached tokens to matview; allow matview refersh on the fly) } dialect := s.db.Dialector.Name() baseQuery := s.ScopedDB(ctx).Model(&Log{}) @@ -3506,7 +3609,9 @@ func (s *RDBLogStore) applyLikeFilter(q *gorm.DB, column, search string) *gorm.D // Scoped to recent data to avoid full table scans. func (s *RDBLogStore) GetDistinctModels(ctx context.Context, limit int, query string) ([]string, error) { if s.db.Dialector.Name() == "postgres" && s.matViewsReady.Load() { - return s.getDistinctModelsFromMatView(ctx, limit, query) + if res, err := s.getDistinctModelsFromMatView(ctx, limit, query); !s.fallBackToRaw(err) { + return res, err + } } cutoff := time.Now().UTC().AddDate(0, 0, -defaultFilterDataCutoffDays) var models []string @@ -3528,7 +3633,9 @@ func (s *RDBLogStore) GetDistinctModels(ctx context.Context, limit int, query st // Scoped to recent data to avoid full table scans. func (s *RDBLogStore) GetDistinctAliases(ctx context.Context, limit int, query string) ([]string, error) { if s.db.Dialector.Name() == "postgres" && s.matViewsReady.Load() { - return s.getDistinctAliasesFromMatView(ctx, limit, query) + if res, err := s.getDistinctAliasesFromMatView(ctx, limit, query); !s.fallBackToRaw(err) { + return res, err + } } cutoff := time.Now().UTC().AddDate(0, 0, -defaultFilterDataCutoffDays) var aliases []string @@ -3574,7 +3681,7 @@ var allowedKeyPairColumns = map[string]struct{}{ func (s *RDBLogStore) GetDistinctKeyPairs(ctx context.Context, idCol, nameCol string, limit int, query string) ([]KeyPairResult, error) { if s.db.Dialector.Name() == "postgres" && s.matViewsReady.Load() { results, served, err := s.getDistinctKeyPairsFromMatView(ctx, idCol, nameCol, limit, query) - if served { + if served && !s.fallBackToRaw(err) { return results, err } } @@ -3604,7 +3711,9 @@ func (s *RDBLogStore) GetDistinctKeyPairs(ctx context.Context, idCol, nameCol st // Scoped to recent data to avoid full table scans. func (s *RDBLogStore) GetDistinctRoutingEngines(ctx context.Context, limit int, query string) ([]string, error) { if s.db.Dialector.Name() == "postgres" && s.matViewsReady.Load() { - return s.getDistinctRoutingEnginesFromMatView(ctx, limit, query) + if res, err := s.getDistinctRoutingEnginesFromMatView(ctx, limit, query); !s.fallBackToRaw(err) { + return res, err + } } cutoff := time.Now().UTC().AddDate(0, 0, -defaultFilterDataCutoffDays) var rawValues []string @@ -3644,7 +3753,9 @@ func (s *RDBLogStore) GetDistinctRoutingEngines(ctx context.Context, limit int, // Scoped to recent data to avoid full table scans. func (s *RDBLogStore) GetDistinctStopReasons(ctx context.Context, limit int, query string) ([]string, error) { if s.db.Dialector.Name() == "postgres" && s.matViewsReady.Load() { - return s.getDistinctStopReasonsFromMatView(ctx, limit, query) + if res, err := s.getDistinctStopReasonsFromMatView(ctx, limit, query); !s.fallBackToRaw(err) { + return res, err + } } cutoff := time.Now().UTC().AddDate(0, 0, -defaultFilterDataCutoffDays) var stopReasons []string diff --git a/framework/logstore/tables.go b/framework/logstore/tables.go index 0e97f8a748d..b3ea19cfdb9 100644 --- a/framework/logstore/tables.go +++ b/framework/logstore/tables.go @@ -1,6 +1,7 @@ package logstore import ( + "database/sql/driver" "strings" "time" @@ -1199,6 +1200,12 @@ const ( WebhookDeliveryOutcomeExhausted WebhookDeliveryOutcome = "exhausted" ) +// Value implements driver.Valuer so database drivers that append typed column +// values (e.g. clickhouse-go batch inserts) can serialize the type. +func (o WebhookDeliveryOutcome) Value() (driver.Value, error) { + return string(o), nil +} + // WebhookDelivery records one webhook delivery attempt. Rows are insert-only // — every attempt appends a new record and existing rows are never updated — // and carry delivery metadata only, never event payloads or receiver