diff --git a/backend/internal/repository/channel_monitor_v2_aggregation.go b/backend/internal/repository/channel_monitor_v2_aggregation.go index 9bec5877b5..eb3d489c6e 100644 --- a/backend/internal/repository/channel_monitor_v2_aggregation.go +++ b/backend/internal/repository/channel_monitor_v2_aggregation.go @@ -395,11 +395,21 @@ func sameFixedRollupBucket(start, end time.Time, seconds int) bool { return start.Truncate(interval).Equal(end.Add(-time.Nanosecond).Truncate(interval)) } +// PostgreSQL interprets a TIMESTAMPTZ literal without an explicit offset in +// the current session timezone. Keep date_bin's origin fixed in UTC so bucket +// boundaries do not shift when the database session runs in Asia/Shanghai (or +// any other non-UTC timezone). +const channelMonitorV2DateBinOrigin = "TIMESTAMPTZ '1970-01-01 00:00:00+00'" + +func channelMonitorV2DateBinExpr(column string) string { + return "date_bin($1::interval," + column + "," + channelMonitorV2DateBinOrigin + ")" +} + const channelMonitorV2FixedRollupBoundsSQL = ` WITH bounds AS ( SELECT - date_bin($1::interval, $3::timestamptz, TIMESTAMPTZ '1970-01-01') AS start_at, - date_bin($1::interval, $4::timestamptz - INTERVAL '1 microsecond', TIMESTAMPTZ '1970-01-01') + $1::interval AS end_at + date_bin($1::interval, $3::timestamptz, ` + channelMonitorV2DateBinOrigin + `) AS start_at, + date_bin($1::interval, $4::timestamptz - INTERVAL '1 microsecond', ` + channelMonitorV2DateBinOrigin + `) + $1::interval AS end_at )` const channelMonitorV2FixedRollupDeleteSQL = channelMonitorV2FixedRollupBoundsSQL + ` @@ -417,7 +427,7 @@ INSERT INTO channel_monitor_v2_metrics_rollup ( duration_count, computed_at ) ` + channelMonitorV2FixedRollupBoundsSQL + ` -SELECT date_bin($1::interval, m.bucket_start, TIMESTAMPTZ '1970-01-01'), $2::integer, +SELECT date_bin($1::interval, m.bucket_start, ` + channelMonitorV2DateBinOrigin + `), $2::integer, platform, group_id, model, SUM(success_requests), SUM(error_requests), SUM(upstream_affected_requests), SUM(upstream_attempt_count), SUM(input_tokens), SUM(output_tokens), SUM(cache_creation_tokens), SUM(cache_read_tokens), @@ -433,7 +443,7 @@ INSERT INTO channel_monitor_v2_user_metrics_rollup ( ttft_sum_ms, ttft_count, duration_sum_ms, duration_count, computed_at ) ` + channelMonitorV2FixedRollupBoundsSQL + ` -SELECT date_bin($1::interval, m.bucket_start, TIMESTAMPTZ '1970-01-01'), $2::integer, +SELECT date_bin($1::interval, m.bucket_start, ` + channelMonitorV2DateBinOrigin + `), $2::integer, platform, group_id, model, user_id, SUM(success_requests), SUM(error_requests), SUM(input_tokens), SUM(output_tokens), SUM(cache_creation_tokens), SUM(cache_read_tokens), SUM(ttft_sum_ms), SUM(ttft_count), SUM(duration_sum_ms), SUM(duration_count), NOW() @@ -446,7 +456,7 @@ INSERT INTO channel_monitor_v2_latency_histograms_rollup ( bucket_start, bucket_seconds, platform, group_id, model, user_id, metric, upper_bound_ms, sample_count ) ` + channelMonitorV2FixedRollupBoundsSQL + ` -SELECT date_bin($1::interval, h.bucket_start, TIMESTAMPTZ '1970-01-01'), $2::integer, +SELECT date_bin($1::interval, h.bucket_start, ` + channelMonitorV2DateBinOrigin + `), $2::integer, platform, group_id, model, user_id, metric, upper_bound_ms, SUM(sample_count) FROM channel_monitor_v2_latency_histograms_1m h, bounds WHERE h.bucket_start >= bounds.start_at AND h.bucket_start < bounds.end_at @@ -457,7 +467,7 @@ INSERT INTO channel_monitor_v2_error_metrics_rollup ( bucket_start, bucket_seconds, platform, group_id, model, error_category, taxonomy_version, error_requests ) ` + channelMonitorV2FixedRollupBoundsSQL + ` -SELECT date_bin($1::interval, e.bucket_start, TIMESTAMPTZ '1970-01-01'), $2::integer, +SELECT date_bin($1::interval, e.bucket_start, ` + channelMonitorV2DateBinOrigin + `), $2::integer, platform, group_id, model, error_category, taxonomy_version, SUM(error_requests) FROM channel_monitor_v2_error_metrics_1m e, bounds WHERE e.bucket_start >= bounds.start_at AND e.bucket_start < bounds.end_at diff --git a/backend/internal/repository/channel_monitor_v2_repo.go b/backend/internal/repository/channel_monitor_v2_repo.go index 978fb44bb0..025f4df8f1 100644 --- a/backend/internal/repository/channel_monitor_v2_repo.go +++ b/backend/internal/repository/channel_monitor_v2_repo.go @@ -949,7 +949,7 @@ func (r *channelMonitorV2Repository) loadFacts(ctx context.Context, filter servi } else { args = append([]any{fmt.Sprintf("%d seconds", int(filter.Bucket.Seconds()))}, args...) where = shiftSQLPlaceholders(where, 1) - bucketExpr = "date_bin($1::interval,m.bucket_start,TIMESTAMPTZ '1970-01-01')" + bucketExpr = channelMonitorV2DateBinExpr("m.bucket_start") group = bucketExpr + "," + group } } @@ -991,7 +991,7 @@ func (r *channelMonitorV2Repository) loadHistograms(ctx context.Context, filter args = []any{fmt.Sprintf("%d seconds", int(filter.Bucket.Seconds()))} args = append(args, oldArgs...) where = shiftSQLPlaceholders(where, 1) - bucketExpr = "date_bin($1::interval,h.bucket_start,TIMESTAMPTZ '1970-01-01')" + bucketExpr = channelMonitorV2DateBinExpr("h.bucket_start") group = bucketExpr + "," + group } } @@ -1444,7 +1444,7 @@ func (r *channelMonitorV2Repository) loadIgnoredErrorCounts( if filter.Bucket > 0 && bucketSeconds == 0 { args = append([]any{fmt.Sprintf("%d seconds", int(filter.Bucket.Seconds()))}, args...) where = shiftSQLPlaceholders(where, 1) - bucketExpr = "date_bin($1::interval,e.bucket_start,TIMESTAMPTZ '1970-01-01')" + bucketExpr = channelMonitorV2DateBinExpr("e.bucket_start") groupBy = bucketExpr + ", e.platform, e.model" } args = append(args, pq.Array(cfg.IgnoredErrorCategories), service.ChannelMonitorV2TaxonomyVersion) @@ -1540,7 +1540,7 @@ func (r *channelMonitorV2Repository) loadIgnoredErrorCountsByMatrixKey( if filter.Bucket > 0 && bucketSeconds == 0 { args = append([]any{fmt.Sprintf("%d seconds", int(filter.Bucket.Seconds()))}, args...) where = shiftSQLPlaceholders(where, 1) - bucketExpr = "date_bin($1::interval,e.bucket_start,TIMESTAMPTZ '1970-01-01')" + bucketExpr = channelMonitorV2DateBinExpr("e.bucket_start") groupSQL = bucketExpr + ", e.platform, e.group_id, e.model" } args = append(args, pq.Array(cfg.IgnoredErrorCategories), service.ChannelMonitorV2TaxonomyVersion) diff --git a/backend/internal/repository/channel_monitor_v2_repo_test.go b/backend/internal/repository/channel_monitor_v2_repo_test.go index c6cec9c926..ac3ae7317c 100644 --- a/backend/internal/repository/channel_monitor_v2_repo_test.go +++ b/backend/internal/repository/channel_monitor_v2_repo_test.go @@ -12,6 +12,22 @@ import ( "github.com/stretchr/testify/require" ) +func TestChannelMonitorV2DateBinOriginIsUTC(t *testing.T) { + require.Equal(t, "TIMESTAMPTZ '1970-01-01 00:00:00+00'", channelMonitorV2DateBinOrigin) + require.Equal(t, "date_bin($1::interval,m.bucket_start,TIMESTAMPTZ '1970-01-01 00:00:00+00')", channelMonitorV2DateBinExpr("m.bucket_start")) + + for _, query := range []string{ + channelMonitorV2FixedRollupBoundsSQL, + channelMonitorV2MetricsRollupSQL, + channelMonitorV2UserMetricsRollupSQL, + channelMonitorV2HistogramRollupSQL, + channelMonitorV2ErrorRollupSQL, + } { + require.Contains(t, query, channelMonitorV2DateBinOrigin) + require.NotContains(t, query, "TIMESTAMPTZ '1970-01-01'") + } +} + func TestChannelMonitorV2DisplayModelIsPlatformScoped(t *testing.T) { cfg := service.ChannelMonitorV2Config{Platforms: []service.ChannelMonitorV2PlatformConfig{ {Platform: "openai", Enabled: true, Models: []string{"shared", "gpt-5"}},