Use cached timestamps in metrics aggreagtion query

runtime/metricsview/executor.go

@@ -155,9 +155,10 @@ func (e *Executor) Timestamps(ctx context.Context, executionTime *time.Time) (ti
 	return e.min, e.max, e.watermark, nil
 }
 
-func (e *Executor) BindQuery(ctx context.Context, qry *Query, min, max, watermark time.Time) error {
-	e.min = min
-	e.max = max
+// BindQuery allows to set min, max and watermark from a cache.
+func (e *Executor) BindQuery(ctx context.Context, qry *Query, minTime, maxTime, watermark time.Time) error {
+	e.min = minTime
+	e.max = maxTime
 	e.watermark = watermark
 	return e.rewriteQueryTimeRanges(ctx, qry, nil)
 }

runtime/metricsview/executor_rewrite_time.go

@@ -12,7 +12,7 @@ import (
 
 // rewriteQueryTimeRanges rewrites the time ranges in the query to fixed start/end timestamps.
 func (e *Executor) rewriteQueryTimeRanges(ctx context.Context, qry *Query, executionTime *time.Time) error {
-	minTime, maxTime, watermark, err := e.Timestamps(ctx, executionTime)
+	_, _, watermark, err := e.Timestamps(ctx, executionTime)
 	if err != nil {
 		return fmt.Errorf("failed to fetch time stamps: %w", err)
 	}
@@ -26,12 +26,12 @@ func (e *Executor) rewriteQueryTimeRanges(ctx context.Context, qry *Query, execu
 		}
 	}
 
-	err = e.resolveTimeRange(ctx, qry.TimeRange, tz, minTime, maxTime, watermark)
+	err = e.resolveTimeRange(qry.TimeRange, tz, watermark)
 	if err != nil {
 		return fmt.Errorf("failed to resolve time range: %w", err)
 	}
 
-	err = e.resolveTimeRange(ctx, qry.ComparisonTimeRange, tz, minTime, maxTime, watermark)
+	err = e.resolveTimeRange(qry.ComparisonTimeRange, tz, watermark)
 	if err != nil {
 		return fmt.Errorf("failed to resolve comparison time range: %w", err)
 	}
@@ -40,7 +40,7 @@ func (e *Executor) rewriteQueryTimeRanges(ctx context.Context, qry *Query, execu
 }
 
 // resolveTimeRange resolves the given time range, ensuring only its Start and End properties are populated.
-func (e *Executor) resolveTimeRange(ctx context.Context, tr *TimeRange, tz *time.Location, minTime, maxTime, watermark time.Time) error {
+func (e *Executor) resolveTimeRange(tr *TimeRange, tz *time.Location, watermark time.Time) error {
 	if tr == nil || tr.IsZero() {
 		return nil
 	}

runtime/queries/metricsview_aggregation.go

@@ -3,10 +3,12 @@ package queries
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
 	"strings"
+	"time"
 
 	runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1"
 	"github.com/rilldata/rill/runtime"
@@ -130,6 +132,48 @@ func (q *MetricsViewAggregation) Export(ctx context.Context, rt *runtime.Runtime
 	}
 	defer e.Close()
 
+	res, err := rt.Resolve(ctx, &runtime.ResolveOptions{
+		InstanceID: instanceID,
+		Resolver:   "metrics_time_range",
+		ResolverProperties: map[string]any{
+			"metrics_view": q.MetricsViewName,
+		},
+		Args: map[string]any{
+			"priority": q.Priority,
+		},
+		Claims: q.SecurityClaims,
+	})
+	if err != nil {
+		return err
+	}
+	defer res.Close()
+
+	row, err := res.Next()
+	if err != nil {
+		if errors.Is(err, io.EOF) {
+			return errors.New("time range query returned no results")
+		}
+		return err
+	}
+
+	minTime, err := anyToTime(row["min"].(string))
+	if err != nil {
+		return err
+	}
+	maxTime, err := anyToTime(row["max"].(string))
+	if err != nil {
+		return err
+	}
+	watermark, err := anyToTime(row["watermark"].(string))
+	if err != nil {
+		return err
+	}
+
+	err = e.BindQuery(ctx, qry, minTime, maxTime, watermark)
+	if err != nil {
+		return err
+	}
+
 	var format drivers.FileFormat
 	switch opts.Format {
 	case runtimev1.ExportFormat_EXPORT_FORMAT_CSV:
@@ -355,3 +399,11 @@ func metricViewExpression(expr *runtimev1.Expression, sql string) (*metricsview.
 	}
 	return nil, nil
 }
+
+func anyToTime(tm any) (time.Time, error) {
+	tmStr, ok := tm.(string)
+	if !ok {
+		return time.Time{}, errors.New("invalid type")
+	}
+	return time.Parse(time.RFC3339, tmStr)
+}

runtime/queries/metricsview_aggregation_test.go

@@ -63,6 +63,9 @@ func TestMetricViewAggregationAgainstClickHouse(t *testing.T) {
 	t.Run("testMetricsViewsAggregation_comparison_with_offset_and_limit_and_delta", func(t *testing.T) {
 		testMetricsViewsAggregation_comparison_with_offset_and_limit_and_delta(t, rt, instanceID)
 	})
+	t.Run("testMetricsViewsAggregation_comparison_iso_timerange", func(t *testing.T) {
+		testMetricsViewsAggregation_comparison_iso_timerange(t, rt, instanceID)
+	})
 }
 
 func TestMetricViewAggregationAgainstDuckDB(t *testing.T) {