From aab6b0a4217bafc545aea44c5bf1108d9bc59445 Mon Sep 17 00:00:00 2001 From: Ian Date: Wed, 26 Aug 2026 21:48:40 +0800 Subject: [PATCH] fix(runtime): terminate fill-missing time spines for unaligned bounds Generate a finite half-open bin list in timeutil, hard-capped at 1500, so looping OLAP dialects cannot hang or allocate unbounded SQL when start and end are not grain-aligned. --- runtime/drivers/clickhouse/dialect.go | 10 +- runtime/drivers/druid/dialect.go | 10 +- runtime/drivers/pinot/dialect.go | 10 +- runtime/drivers/snowflake/dialect.go | 12 +- runtime/drivers/starrocks/dialect.go | 12 +- runtime/drivers/starrocks/starrocks_test.go | 28 +++++ runtime/pkg/timeutil/timeutil.go | 27 ++++ runtime/pkg/timeutil/timeutil_test.go | 133 ++++++++++++++++++++ 8 files changed, 218 insertions(+), 24 deletions(-) diff --git a/runtime/drivers/clickhouse/dialect.go b/runtime/drivers/clickhouse/dialect.go index b32f9622624b..8af3be29a258 100644 --- a/runtime/drivers/clickhouse/dialect.go +++ b/runtime/drivers/clickhouse/dialect.go @@ -149,14 +149,16 @@ func (d *dialect) IntervalSubtract(tsExpr, unitExpr string, grain runtimev1.Time } func (d *dialect) SelectTimeRangeBins(start, end time.Time, grain runtimev1.TimeGrain, alias string, tz *time.Location, firstDay, firstMonth int) (string, []any, error) { - g := timeutil.TimeGrainFromAPI(grain) - start = timeutil.TruncateTime(start, g, tz, firstDay, firstMonth) + bins, err := timeutil.TimeRangeBins(start, end, timeutil.TimeGrainFromAPI(grain), tz, firstDay, firstMonth) + if err != nil { + return "", nil, err + } // format: SELECT c1 AS "alias" FROM VALUES(toDateTime(...), ...) var sb strings.Builder var args []any sb.WriteString(fmt.Sprintf("SELECT c1 AS %s FROM VALUES(", d.EscapeAlias(alias))) - for t := start; t.Before(end); t = timeutil.OffsetTime(t, g, 1, tz) { - if t != start { + for i, t := range bins { + if i > 0 { sb.WriteString(", ") } sb.WriteString("?") diff --git a/runtime/drivers/druid/dialect.go b/runtime/drivers/druid/dialect.go index 68ba2a2a6f60..6967e412730a 100644 --- a/runtime/drivers/druid/dialect.go +++ b/runtime/drivers/druid/dialect.go @@ -115,8 +115,10 @@ func (d *dialect) IntervalSubtract(tsExpr, unitExpr string, grain runtimev1.Time } func (d *dialect) SelectTimeRangeBins(start, end time.Time, grain runtimev1.TimeGrain, alias string, tz *time.Location, firstDay, firstMonth int) (string, []any, error) { - g := timeutil.TimeGrainFromAPI(grain) - start = timeutil.TruncateTime(start, g, tz, firstDay, firstMonth) + bins, err := timeutil.TimeRangeBins(start, end, timeutil.TimeGrainFromAPI(grain), tz, firstDay, firstMonth) + if err != nil { + return "", nil, err + } // generate select like - SELECT * FROM ( // VALUES // (CAST('2006-01-02T15:04:05Z' AS TIMESTAMP)), @@ -125,8 +127,8 @@ func (d *dialect) SelectTimeRangeBins(start, end time.Time, grain runtimev1.Time var sb strings.Builder var args []any sb.WriteString("SELECT * FROM (VALUES ") - for t := start; t.Before(end); t = timeutil.OffsetTime(t, g, 1, tz) { - if t != start { + for i, t := range bins { + if i > 0 { sb.WriteString(", ") } sb.WriteString("(CAST(? AS TIMESTAMP))") diff --git a/runtime/drivers/pinot/dialect.go b/runtime/drivers/pinot/dialect.go index bed2a6b536e0..99e255aa2922 100644 --- a/runtime/drivers/pinot/dialect.go +++ b/runtime/drivers/pinot/dialect.go @@ -77,8 +77,10 @@ func (d *dialect) IntervalSubtract(tsExpr, unitExpr string, grain runtimev1.Time } func (d *dialect) SelectTimeRangeBins(start, end time.Time, grain runtimev1.TimeGrain, alias string, tz *time.Location, firstDay, firstMonth int) (string, []any, error) { - g := timeutil.TimeGrainFromAPI(grain) - start = timeutil.TruncateTime(start, g, tz, firstDay, firstMonth) + bins, err := timeutil.TimeRangeBins(start, end, timeutil.TimeGrainFromAPI(grain), tz, firstDay, firstMonth) + if err != nil { + return "", nil, err + } // generate select like - SELECT * FROM ( // VALUES // (CAST('2006-01-02T15:04:05Z' AS TIMESTAMP)), @@ -87,8 +89,8 @@ func (d *dialect) SelectTimeRangeBins(start, end time.Time, grain runtimev1.Time var sb strings.Builder var args []any sb.WriteString("SELECT * FROM (VALUES ") - for t := start; t.Before(end); t = timeutil.OffsetTime(t, g, 1, tz) { - if t != start { + for i, t := range bins { + if i > 0 { sb.WriteString(", ") } sb.WriteString("(CAST(? AS TIMESTAMP))") diff --git a/runtime/drivers/snowflake/dialect.go b/runtime/drivers/snowflake/dialect.go index a4174c45e59b..28d0473a8ea0 100644 --- a/runtime/drivers/snowflake/dialect.go +++ b/runtime/drivers/snowflake/dialect.go @@ -96,17 +96,17 @@ func (d *dialect) IntervalSubtract(tsExpr, unitExpr string, grain runtimev1.Time } func (d *dialect) SelectTimeRangeBins(start, end time.Time, grain runtimev1.TimeGrain, alias string, tz *time.Location, firstDay, firstMonth int) (string, []any, error) { - g := timeutil.TimeGrainFromAPI(grain) - start = timeutil.TruncateTime(start, g, tz, firstDay, firstMonth) + bins, err := timeutil.TimeRangeBins(start, end, timeutil.TimeGrainFromAPI(grain), tz, firstDay, firstMonth) + if err != nil { + return "", nil, err + } // Snowflake uses UNION ALL for generating time series var sb strings.Builder - first := true - for t := start; t.Before(end); t = timeutil.OffsetTime(t, g, 1, tz) { - if !first { + for i, t := range bins { + if i > 0 { sb.WriteString(" UNION ALL ") } fmt.Fprintf(&sb, "SELECT CAST('%s' AS TIMESTAMP) AS %s", t.Format(time.RFC3339), d.EscapeAlias(alias)) - first = false } return sb.String(), nil, nil } diff --git a/runtime/drivers/starrocks/dialect.go b/runtime/drivers/starrocks/dialect.go index 30368c100b6f..55e9fce6c804 100644 --- a/runtime/drivers/starrocks/dialect.go +++ b/runtime/drivers/starrocks/dialect.go @@ -84,17 +84,17 @@ func (d *dialect) IntervalSubtract(tsExpr, unitExpr string, grain runtimev1.Time } func (d *dialect) SelectTimeRangeBins(start, end time.Time, grain runtimev1.TimeGrain, alias string, tz *time.Location, firstDay, firstMonth int) (string, []any, error) { - g := timeutil.TimeGrainFromAPI(grain) - start = timeutil.TruncateTime(start, g, tz, firstDay, firstMonth) + bins, err := timeutil.TimeRangeBins(start, end, timeutil.TimeGrainFromAPI(grain), tz, firstDay, firstMonth) + if err != nil { + return "", nil, err + } // StarRocks uses UNION ALL for generating time series. var sb strings.Builder - first := true - for t := start; t != end; t = timeutil.OffsetTime(t, g, 1, tz) { - if !first { + for i, t := range bins { + if i > 0 { sb.WriteString(" UNION ALL ") } sb.WriteString(fmt.Sprintf("SELECT CAST('%s' AS DATETIME) AS %s", t.Format(time.DateTime), d.EscapeAlias(alias))) - first = false } return sb.String(), nil, nil } diff --git a/runtime/drivers/starrocks/starrocks_test.go b/runtime/drivers/starrocks/starrocks_test.go index 8781fff20950..6004b57cae78 100644 --- a/runtime/drivers/starrocks/starrocks_test.go +++ b/runtime/drivers/starrocks/starrocks_test.go @@ -1,8 +1,11 @@ package starrocks import ( + "strings" "testing" + "time" + runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1" "github.com/stretchr/testify/require" ) @@ -195,3 +198,28 @@ func TestDatabaseTypeToRuntimeType(t *testing.T) { }) } } + +func TestSelectTimeRangeBins_unalignedUTCDayRepro(t *testing.T) { + start, err := time.Parse(time.RFC3339, "2026-07-29T16:00:00Z") + require.NoError(t, err) + end, err := time.Parse(time.RFC3339, "2026-08-26T16:00:00Z") + require.NoError(t, err) + + done := make(chan struct{}) + var sql string + var selErr error + go func() { + defer close(done) + sql, _, selErr = DialectStarRocks.SelectTimeRangeBins(start, end, runtimev1.TimeGrain_TIME_GRAIN_DAY, "ts", time.UTC, 1, 1) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("SelectTimeRangeBins hung on unaligned UTC day bounds") + } + + require.NoError(t, selErr) + require.Contains(t, sql, "2026-07-29 00:00:00") + require.Contains(t, sql, "2026-08-26 00:00:00") + require.Equal(t, 28, strings.Count(sql, "UNION ALL")) +} diff --git a/runtime/pkg/timeutil/timeutil.go b/runtime/pkg/timeutil/timeutil.go index 8bf95f790cc1..152bcd6b3ed9 100644 --- a/runtime/pkg/timeutil/timeutil.go +++ b/runtime/pkg/timeutil/timeutil.go @@ -1,6 +1,7 @@ package timeutil import ( + "fmt" "time" runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1" @@ -9,6 +10,10 @@ import ( _ "time/tzdata" ) +// MaxTimeRangeBins is the maximum number of timestamps a fill-missing time spine may contain. +// It matches the ApproximateBins pre-check in metricsview/ast.go. +const MaxTimeRangeBins = 1500 + // TimeGrain is extension of std time package with Week and Quarter added type TimeGrain int @@ -184,6 +189,28 @@ func ApproximateBins(start, end time.Time, tg TimeGrain) int { return -1 } +// TimeRangeBins returns the half-open sequence of truncated timestamps in [start, end). +// Start is truncated to grain in tz. Iteration uses t.Before(end) so unaligned bounds terminate. +// More than MaxTimeRangeBins timestamps is an error; the function never loops unbounded. +func TimeRangeBins(start, end time.Time, tg TimeGrain, tz *time.Location, firstDay, firstMonth int) ([]time.Time, error) { + if tz == nil { + tz = time.UTC + } + if tg == TimeGrainUnspecified { + return nil, fmt.Errorf("time grain is unspecified") + } + + start = TruncateTime(start, tg, tz, firstDay, firstMonth) + var bins []time.Time + for t := start; t.Before(end); t = OffsetTime(t, tg, 1, tz) { + if len(bins) >= MaxTimeRangeBins { + return nil, fmt.Errorf("time range has more than %d bins for %q grain, move to a larger grain", MaxTimeRangeBins, TimeGrainToAPI(tg)) + } + bins = append(bins, t) + } + return bins, nil +} + func OffsetTime(tm time.Time, tg TimeGrain, n int, tz *time.Location) time.Time { tm = tm.In(tz) diff --git a/runtime/pkg/timeutil/timeutil_test.go b/runtime/pkg/timeutil/timeutil_test.go index b9a1981abd7d..a2fe02f47a9e 100644 --- a/runtime/pkg/timeutil/timeutil_test.go +++ b/runtime/pkg/timeutil/timeutil_test.go @@ -147,6 +147,139 @@ func TestOffsetTime(t *testing.T) { require.Equal(t, parseTestTime(t, "2024-03-11T04:00:00Z"), OffsetTime(parseTestTime(t, "2024-03-09T05:00:00Z"), TimeGrainDay, 2, tz)) } +func TestTimeRangeBins_unalignedUTCDayRepro(t *testing.T) { + // Parked hang: start truncated to 00:00Z never equals end 16:00Z under t != end. + start := parseTestTime(t, "2026-07-29T16:00:00Z") + end := parseTestTime(t, "2026-08-26T16:00:00Z") + + bins, err := TimeRangeBins(start, end, TimeGrainDay, time.UTC, 1, 1) + require.NoError(t, err) + require.NotEmpty(t, bins) + require.LessOrEqual(t, len(bins), MaxTimeRangeBins) + + require.Equal(t, parseTestTime(t, "2026-07-29T00:00:00Z"), bins[0]) + require.Equal(t, parseTestTime(t, "2026-08-26T00:00:00Z"), bins[len(bins)-1]) + require.Equal(t, TruncateTime(start, TimeGrainDay, time.UTC, 1, 1), bins[0]) + + assertDateTruncAlignedSpine(t, bins, start, end, TimeGrainDay, time.UTC, 1, 1) +} + +func TestTimeRangeBins_exactEndEquality(t *testing.T) { + t.Run("truncated start equals end", func(t *testing.T) { + start := parseTestTime(t, "2026-08-26T00:00:00Z") + end := parseTestTime(t, "2026-08-26T00:00:00Z") + bins, err := TimeRangeBins(start, end, TimeGrainDay, time.UTC, 1, 1) + require.NoError(t, err) + require.Empty(t, bins) + }) + + t.Run("aligned bounds exclude exclusive end", func(t *testing.T) { + start := parseTestTime(t, "2026-07-29T00:00:00Z") + end := parseTestTime(t, "2026-08-26T00:00:00Z") + bins, err := TimeRangeBins(start, end, TimeGrainDay, time.UTC, 1, 1) + require.NoError(t, err) + require.Equal(t, start, bins[0]) + require.Equal(t, parseTestTime(t, "2026-08-25T00:00:00Z"), bins[len(bins)-1]) + for _, b := range bins { + require.False(t, b.Equal(end)) + require.True(t, b.Before(end)) + } + assertDateTruncAlignedSpine(t, bins, start, end, TimeGrainDay, time.UTC, 1, 1) + }) +} + +func TestTimeRangeBins_emptyRange(t *testing.T) { + start := parseTestTime(t, "2026-08-26T16:00:00Z") + end := parseTestTime(t, "2026-07-29T16:00:00Z") + bins, err := TimeRangeBins(start, end, TimeGrainDay, time.UTC, 1, 1) + require.NoError(t, err) + require.Empty(t, bins) +} + +func TestTimeRangeBins_over1500Errors(t *testing.T) { + start := parseTestTime(t, "2020-01-01T00:00:00Z") + + t.Run("exactly 1500 bins succeeds", func(t *testing.T) { + end := start.Add(1500 * time.Hour) + bins, err := TimeRangeBins(start, end, TimeGrainHour, time.UTC, 1, 1) + require.NoError(t, err) + require.Len(t, bins, MaxTimeRangeBins) + }) + + t.Run("1501 hours errors", func(t *testing.T) { + end := start.Add(1501 * time.Hour) + type outcome struct { + bins []time.Time + err error + } + done := make(chan outcome, 1) + go func() { + bins, err := TimeRangeBins(start, end, TimeGrainHour, time.UTC, 1, 1) + done <- outcome{bins, err} + }() + select { + case got := <-done: + require.Error(t, got.err) + require.Nil(t, got.bins) + require.Contains(t, got.err.Error(), "1500") + case <-time.After(2 * time.Second): + t.Fatal("TimeRangeBins hung instead of erroring above 1500 bins") + } + }) + + t.Run("millisecond grain over a day errors without hanging", func(t *testing.T) { + end := start.Add(24 * time.Hour) + type outcome struct { + bins []time.Time + err error + } + done := make(chan outcome, 1) + go func() { + bins, err := TimeRangeBins(start, end, TimeGrainMillisecond, time.UTC, 1, 1) + done <- outcome{bins, err} + }() + select { + case got := <-done: + require.Error(t, got.err) + require.Nil(t, got.bins) + require.Contains(t, got.err.Error(), "1500") + case <-time.After(2 * time.Second): + t.Fatal("TimeRangeBins hung instead of erroring above 1500 bins") + } + }) +} + +func assertDateTruncAlignedSpine(t *testing.T, bins []time.Time, start, end time.Time, tg TimeGrain, tz *time.Location, firstDay, firstMonth int) { + t.Helper() + require.NotEmpty(t, bins) + require.Equal(t, TruncateTime(start, tg, tz, firstDay, firstMonth), bins[0], "first bin must be date_trunc of the range start") + require.Equal(t, TruncateTime(end.Add(-time.Nanosecond), tg, tz, firstDay, firstMonth), bins[len(bins)-1], "last bin must be date_trunc of the last instant in [start, end)") + + seen := make(map[int64]struct{}, len(bins)) + for i, b := range bins { + // Each spine value is already date_trunc'd, matching the aggregation GROUP BY bucket. + require.Equal(t, TruncateTime(b, tg, tz, firstDay, firstMonth), b, "bin %s is not date_trunc aligned", b.Format(time.RFC3339)) + require.True(t, b.Before(end), "bin %s is not before exclusive end", b.Format(time.RFC3339)) + if i > 0 { + require.Equal(t, OffsetTime(bins[i-1], tg, 1, tz), b, "bin %d is not one grain after the previous", i) + } + seen[b.UnixNano()] = struct{}{} + } + + samples := []time.Time{start, end.Add(-time.Nanosecond)} + if end.Sub(start) > 2*time.Hour { + samples = append(samples, start.Add(time.Hour), start.Add(end.Sub(start)/2)) + } + for _, ts := range samples { + if ts.Before(start) || !ts.Before(end) { + continue + } + truncated := TruncateTime(ts, tg, tz, firstDay, firstMonth) + _, ok := seen[truncated.UnixNano()] + require.True(t, ok, "date_trunc(%s)=%s is not in the spine", ts.Format(time.RFC3339), truncated.Format(time.RFC3339)) + } +} + func parseTestTime(tst *testing.T, t string) time.Time { ts, err := time.Parse(time.RFC3339, t) require.NoError(tst, err)