From e0ff74b5f18475cd30416bfeb10fb891bbd701f6 Mon Sep 17 00:00:00 2001 From: vcarey Date: Thu, 2 Jul 2026 10:54:03 -0400 Subject: [PATCH] Improve cache warmup token and limit handling --- internal/translator/function/cache_warmup.go | 31 ++ .../translator/function/cache_warmup_test.go | 20 + internal/translator/view.go | 36 +- internal/translator/view_warmup_test.go | 32 +- service/reader/sql.go | 3 + service/reader/sql_test.go | 427 +----------------- view/cache.go | 95 +++- view/cache_warmup_test.go | 127 ++++++ view/state.go | 10 +- warmup/cache_test.go | 100 ++++ 10 files changed, 441 insertions(+), 440 deletions(-) create mode 100644 view/cache_warmup_test.go diff --git a/internal/translator/function/cache_warmup.go b/internal/translator/function/cache_warmup.go index 74c2f1813..c5759f0fc 100644 --- a/internal/translator/function/cache_warmup.go +++ b/internal/translator/function/cache_warmup.go @@ -2,6 +2,7 @@ package function import ( "fmt" + "strconv" "strings" "github.com/viant/datly/view" @@ -42,6 +43,13 @@ func (c *cacheWarmup) Apply(args []string, column *sqlparser.Column, resource *v warmup.FieldNames = fieldNames continue } + if limit, ok, err := parseWarmupLimit(raw); ok || err != nil { + if err != nil { + return err + } + warmup.Limit = limit + continue + } param, err := parseWarmupParam(raw) if err != nil { return err @@ -121,6 +129,29 @@ func parseWarmupFieldNames(raw string) ([]string, bool, error) { return result, true, nil } +func parseWarmupLimit(raw string) (*int, bool, error) { + name, value, ok := splitWarmupOption(raw) + if !ok { + return nil, false, nil + } + switch strings.ToLower(name) { + case "limit": + default: + return nil, false, nil + } + if value == "" { + return nil, true, fmt.Errorf("warmup limit was empty") + } + limit, err := strconv.Atoi(value) + if err != nil { + return nil, true, fmt.Errorf("warmup limit %q was invalid: %w", value, err) + } + if limit < 0 { + return nil, true, fmt.Errorf("warmup limit %q must be zero or greater", value) + } + return &limit, true, nil +} + func parseWarmupParam(raw string) (*view.ParamValue, error) { name, rawValues, ok := splitWarmupOption(raw) if !ok { diff --git a/internal/translator/function/cache_warmup_test.go b/internal/translator/function/cache_warmup_test.go index 220e89cc3..a75b5c7f8 100644 --- a/internal/translator/function/cache_warmup_test.go +++ b/internal/translator/function/cache_warmup_test.go @@ -14,6 +14,7 @@ func TestCacheWarmupApply(t *testing.T) { "order_id", "Connector=bq_metrics_prewarm", "IndexParameter=OrderId", + "Limit=0", "FieldNames=Id,Name", "Period=today,yesterday", "Granularity=hour,day", @@ -30,6 +31,9 @@ func TestCacheWarmupApply(t *testing.T) { if aView.Cache.Warmup.IndexParameter != "OrderId" { t.Fatalf("unexpected index parameter: %v", aView.Cache.Warmup.IndexParameter) } + if aView.Cache.Warmup.Limit == nil || *aView.Cache.Warmup.Limit != 0 { + t.Fatalf("unexpected warmup limit: %#v", aView.Cache.Warmup.Limit) + } if len(aView.Cache.Warmup.FieldNames) != 2 || aView.Cache.Warmup.FieldNames[0] != "Id" || aView.Cache.Warmup.FieldNames[1] != "Name" { t.Fatalf("unexpected field names: %#v", aView.Cache.Warmup.FieldNames) } @@ -74,3 +78,19 @@ func TestCacheWarmupApplyRejectsEmptyFieldNames(t *testing.T) { t.Fatalf("expected error") } } + +func TestCacheWarmupApplyRejectsInvalidLimit(t *testing.T) { + aView := &view.View{Cache: view.NewRefCache("aerospike")} + err := (&cacheWarmup{}).Apply([]string{"order_id", "Limit=invalid"}, nil, &view.Resource{}, aView) + if err == nil { + t.Fatalf("expected error") + } +} + +func TestCacheWarmupApplyRejectsNegativeLimit(t *testing.T) { + aView := &view.View{Cache: view.NewRefCache("aerospike")} + err := (&cacheWarmup{}).Apply([]string{"order_id", "Limit=-1"}, nil, &view.Resource{}, aView) + if err == nil { + t.Fatalf("expected error") + } +} diff --git a/internal/translator/view.go b/internal/translator/view.go index 975df5581..b2967515d 100644 --- a/internal/translator/view.go +++ b/internal/translator/view.go @@ -3,6 +3,7 @@ package translator import ( "fmt" "path" + "strconv" "strings" "github.com/viant/datly/internal/asset" @@ -43,11 +44,10 @@ func (v *View) applyHintSettings(namespace *Viewlet) error { return fmt.Errorf("invalid view %v hint, %w, %s", v, err, viewJSONHint) } - v.applyShorthands(namespace) - return nil + return v.applyShorthands(namespace) } -func (v *View) applyShorthands(viewlet *Viewlet) { +func (v *View) applyShorthands(viewlet *Viewlet) error { if v.Self != nil { v.SelfReference = v.Self } @@ -79,13 +79,18 @@ func (v *View) applyShorthands(viewlet *Viewlet) { } if len(v.Warmup) > 0 { - v.View.Cache.Warmup = v.buildCacheWarmup(v.Warmup, viewlet) + warmup, err := v.buildCacheWarmup(v.Warmup, viewlet) + if err != nil { + return err + } + v.View.Cache.Warmup = warmup } + return nil } -func (v *View) buildCacheWarmup(warmup map[string]interface{}, viewlet *Viewlet) *view.Warmup { +func (v *View) buildCacheWarmup(warmup map[string]interface{}, viewlet *Viewlet) (*view.Warmup, error) { if warmup == nil { - return nil + return nil, nil } warmup = copyWarmup(warmup) @@ -93,6 +98,8 @@ func (v *View) buildCacheWarmup(warmup map[string]interface{}, viewlet *Viewlet) delete(warmup, "IndexColumn") indexParameter, _ := warmup["IndexParameter"] delete(warmup, "IndexParameter") + limit, hasLimit := warmup["Limit"] + delete(warmup, "Limit") connector, _ := warmup["Connector"] delete(warmup, "Connector") var refColumn string @@ -107,11 +114,24 @@ func (v *View) buildCacheWarmup(warmup map[string]interface{}, viewlet *Viewlet) result.IndexColumn = explicit } if result.IndexColumn == "" { - return nil + return nil, nil } if parameterName := strings.TrimSpace(fmt.Sprint(indexParameter)); parameterName != "" && parameterName != "" { result.IndexParameter = parameterName } + if hasLimit { + limitValue := strings.TrimSpace(fmt.Sprint(limit)) + if limitValue != "" && limitValue != "" { + parsed, err := strconv.Atoi(limitValue) + if err != nil { + return nil, fmt.Errorf("invalid warmup limit %q: %w", limitValue, err) + } + if parsed < 0 { + return nil, fmt.Errorf("invalid warmup limit %q: must be zero or greater", limitValue) + } + result.Limit = &parsed + } + } if connectorName := strings.TrimSpace(fmt.Sprint(connector)); connectorName != "" && connectorName != "" { result.Connector = view.NewRefConnector(connectorName) } @@ -127,7 +147,7 @@ func (v *View) buildCacheWarmup(warmup map[string]interface{}, viewlet *Viewlet) } result.Cases = append(result.Cases, multiSet) - return result + return result, nil } func copyWarmup(warmup map[string]interface{}) map[string]interface{} { diff --git a/internal/translator/view_warmup_test.go b/internal/translator/view_warmup_test.go index f345adada..085c14ea2 100644 --- a/internal/translator/view_warmup_test.go +++ b/internal/translator/view_warmup_test.go @@ -10,17 +10,47 @@ func TestViewBuildCacheWarmup_RootViewUsesExplicitIndexColumn(t *testing.T) { subject := &View{} viewlet := &Viewlet{Name: "adOrderRoot"} - warmup := subject.buildCacheWarmup(map[string]interface{}{ + warmup, err := subject.buildCacheWarmup(map[string]interface{}{ "IndexColumn": "ad_order_id", "IndexParameter": "AdOrderId", + "Limit": "0", "Connector": "bq_metrics_prewarm", }, viewlet) + require.NoError(t, err) require.NotNil(t, warmup) require.Equal(t, "ad_order_id", warmup.IndexColumn) require.Equal(t, "AdOrderId", warmup.IndexParameter) + require.NotNil(t, warmup.Limit) + require.Equal(t, 0, *warmup.Limit) require.NotNil(t, warmup.Connector) require.Equal(t, "bq_metrics_prewarm", warmup.Connector.Ref) require.Len(t, warmup.Cases, 1) require.Len(t, warmup.Cases[0].Set, 0) } + +func TestViewBuildCacheWarmup_InvalidLimitReturnsError(t *testing.T) { + subject := &View{} + viewlet := &Viewlet{Name: "adOrderRoot"} + + warmup, err := subject.buildCacheWarmup(map[string]interface{}{ + "IndexColumn": "ad_order_id", + "Limit": "invalid", + }, viewlet) + + require.Nil(t, warmup) + require.Error(t, err) +} + +func TestViewBuildCacheWarmup_NegativeLimitReturnsError(t *testing.T) { + subject := &View{} + viewlet := &Viewlet{Name: "adOrderRoot"} + + warmup, err := subject.buildCacheWarmup(map[string]interface{}{ + "IndexColumn": "ad_order_id", + "Limit": "-1", + }, viewlet) + + require.Nil(t, warmup) + require.Error(t, err) +} diff --git a/service/reader/sql.go b/service/reader/sql.go index bbaf298d3..be0d86f65 100644 --- a/service/reader/sql.go +++ b/service/reader/sql.go @@ -911,6 +911,9 @@ func (b *Builder) lookupRelationColumn(aView *view.View, relation *view.Relation } func actualLimit(aView *view.View, selector *view.Statelet) int { + if selector.WarmupNoLimit { + return 0 + } if selector.Limit != 0 { return selector.Limit } diff --git a/service/reader/sql_test.go b/service/reader/sql_test.go index 4d90b967f..0d2692b33 100644 --- a/service/reader/sql_test.go +++ b/service/reader/sql_test.go @@ -1,428 +1,17 @@ package reader -/* import ( - "context" - "fmt" + "testing" + "github.com/stretchr/testify/assert" - "github.com/viant/assertly" - "github.com/viant/datly/internal/tests" "github.com/viant/datly/view" - "github.com/viant/datly/view/state" - "github.com/viant/dsunit" - "github.com/viant/toolbox" - "path" - "reflect" - "strings" - "testing" ) -func TestBuilder_Build(t *testing.T) { - testLocation := toolbox.CallerDirectory(3) - - type Params struct { - EventId int - } - - type PresenceMap struct { - } - - useCases := []struct { - batchData *view.BatchData - view *view.View - relation *view.Relation - selector *view.Statelet - placeholders []interface{} - description string - output string - dataset string - }{ - { - dataset: "dataset001_events/", - description: `select statement`, - output: `SELECT t.ID, t.Price FROM events AS t`, - view: &view.View{ - Columns: []*view.Column{ - { - Name: "ID", - DataType: "Int", - }, - { - Name: "Price", - DataType: "Float", - }, - }, - Name: "events", - Table: "events", - Template: &view.Template{ - Output: state.NewSchema(reflect.TypeOf(Params{})), - PresenceSchema: state.NewSchema(reflect.TypeOf(PresenceMap{})), - }, - }, - batchData: &view.BatchData{}, - selector: &view.Statelet{InputParameters: view.ParamState{ - Values: Params{}, - Has: PresenceMap{}, - }}, - }, - { - dataset: "dataset001_events/", - description: `select statement with offset and limit`, - output: `SELECT t.ID, t.Price FROM events AS t LIMIT 10 OFFSET 5`, - view: &view.View{ - Columns: []*view.Column{ - { - Name: "ID", - DataType: "Int", - }, - { - Name: "Price", - DataType: "Float", - }, - }, - Name: "events", - Selector: &view.Config{ - Limit: 10, - }, - Table: "events", - Template: &view.Template{ - Output: state.NewSchema(reflect.TypeOf(Params{})), - PresenceSchema: state.NewSchema(reflect.TypeOf(PresenceMap{})), - }, - }, - batchData: &view.BatchData{}, - selector: &view.Statelet{ - InputParameters: view.ParamState{ - Values: Params{}, - Has: PresenceMap{}, - }, - Offset: 5, - }, - }, - { - dataset: "dataset001_events/", - description: `select statement with $PAGINATION`, - output: `SELECT t.ID, t.Price FROM (SELECT * FROM EVENTS LIMIT 10 OFFSET 5 ) AS t`, - view: &view.View{ - Columns: []*view.Column{ - { - Name: "ID", - DataType: "Int", - }, - { - Name: "Price", - DataType: "Float", - }, - }, - Name: "events", - Selector: &view.Config{ - Limit: 10, - }, - From: "SELECT * FROM EVENTS $PAGINATION", - Table: "events", - Template: &view.Template{ - Output: state.NewSchema(reflect.TypeOf(Params{})), - PresenceSchema: state.NewSchema(reflect.TypeOf(PresenceMap{})), - }, - }, - batchData: &view.BatchData{}, - selector: &view.Statelet{ - InputParameters: view.ParamState{ - Values: Params{}, - Has: PresenceMap{}, - }, - Offset: 5, - }, - }, - { - dataset: "dataset001_events/", - description: `select statement with View Criteria`, - output: `SELECT t.ID, t.Price FROM (SELECT * FROM EVENTS ) AS t`, - view: &view.View{ - Columns: []*view.Column{ - { - Name: "ID", - DataType: "Int", - }, - { - Name: "Price", - DataType: "Float", - }, - }, - Name: "events", - From: "SELECT * FROM EVENTS $PAGINATION", - Table: "Events", - Template: &view.Template{ - Output: state.NewSchema(reflect.TypeOf(Params{})), - PresenceSchema: state.NewSchema(reflect.TypeOf(PresenceMap{})), - }, - }, - batchData: &view.BatchData{}, - selector: &view.Statelet{ - InputParameters: view.ParamState{ - Values: Params{}, - Has: PresenceMap{}, - }, - }, - }, - { - dataset: "dataset001_events/", - description: `select statement with $WHERE_CRITERIA`, - output: `SELECT t.ID, t.Price FROM (SELECT * FROM EVENTS ) AS t`, - view: &view.View{ - Columns: []*view.Column{ - { - Name: "ID", - DataType: "Int", - }, - { - Name: "Price", - DataType: "Float", - }, - }, - Name: "events", - From: "SELECT * FROM EVENTS $WHERE_CRITERIA", - Table: "Events", - Template: &view.Template{ - Output: state.NewSchema(reflect.TypeOf(Params{})), - PresenceSchema: state.NewSchema(reflect.TypeOf(PresenceMap{})), - }, - }, - batchData: &view.BatchData{}, - selector: &view.Statelet{ - InputParameters: view.ParamState{ - Values: Params{}, - Has: PresenceMap{}, - }, - }, - }, - { - dataset: "dataset001_events/", - description: `select statement with parameters`, - output: `SELECT t.ID, t.Price FROM (SELECT * FROM EVENTS WHERE ID = ? ) AS t`, - view: &view.View{ - Columns: []*view.Column{ - { - Name: "ID", - DataType: "Int", - }, - { - Name: "Price", - DataType: "Float", - }, - }, - Name: "events", - From: "SELECT * FROM EVENTS WHERE ID = $EventId", - Table: "Events", - Template: &view.Template{ - Output: state.NewSchema(reflect.TypeOf(Params{})), - InputParameters: []*state.Parameter{ - { - Name: "EventId", - In: &state.State{ - Kind: state.KindPath, - Name: "eventId", - }, - Output: &state.Output{ - DataType: "int", - }, - }, - }, - }, - }, - placeholders: []interface{}{10}, - batchData: &view.BatchData{}, - selector: &view.Statelet{ - InputParameters: view.ParamState{ - Values: Params{EventId: 10}, - Has: PresenceMap{}, - }, - }, - }, - { - dataset: "dataset001_events/", - description: `select statement with $AND_COLUMN_IN`, - output: `SELECT t.ID, t.Price FROM (SELECT * FROM EVENTS ev WHERE ev.ID = ? AND ( ev.user_id IN (?, ?, ?, ?)) ) AS t`, - placeholders: []interface{}{10, 4, 5, 9, 2}, - relation: &view.Relation{ColumnNamespace: "ev", Of: &view.ReferenceView{Column: "ID"}}, - view: &view.View{ - Columns: []*view.Column{ - { - Name: "ID", - DataType: "Int", - }, - { - Name: "Price", - DataType: "Float", - }, - }, - Name: "events", - From: "SELECT * FROM EVENTS ev WHERE ev.ID = $EventId $AND_COLUMN_IN", - Table: "Events", - Template: &view.Template{ - Output: state.NewSchema(reflect.TypeOf(Params{})), - InputParameters: []*state.Parameter{ - { - Name: "EventId", - In: &state.State{ - Kind: state.KindPath, - Name: "eventId", - }, - Output: &state.Output{ - DataType: "int", - }, - }, - }, - }, - }, - batchData: &view.BatchData{ - ColumnNames: "user_id", - ValuesBatch: []interface{}{4, 5, 9, 2}, - }, - selector: &view.Statelet{ - InputParameters: view.ParamState{ - Values: Params{EventId: 10}, - Has: PresenceMap{}, - }, - }, - }, - { - dataset: "dataset001_events/", - description: `select statement without $COLUMN_IN`, - output: `SELECT t.ID, t.Price FROM (SELECT * FROM EVENTS ev WHERE ev.ID = ? AND ( ev.user_id IN (?, ?, ?, ?)) ) AS t`, - placeholders: []interface{}{10, 4, 5, 9, 2}, - relation: &view.Relation{ColumnNamespace: "ev", Of: &view.ReferenceView{Column: "ID"}}, - view: &view.View{ - Columns: []*view.Column{ - { - Name: "ID", - DataType: "Int", - }, - { - Name: "Price", - DataType: "Float", - }, - }, - Name: "events", - From: "SELECT * FROM EVENTS ev WHERE ev.ID = $EventId", - Table: "Events", - Template: &view.Template{ - Output: state.NewSchema(reflect.TypeOf(Params{})), - InputParameters: []*state.Parameter{ - { - Name: "EventId", - In: &state.State{ - Kind: state.KindPath, - Name: "eventId", - }, - Output: &state.Output{ - DataType: "int", - }, - }, - }, - }, - }, - batchData: &view.BatchData{ - ColumnNames: "user_id", - ValuesBatch: []interface{}{4, 5, 9, 2}, - }, - selector: &view.Statelet{ - InputParameters: view.ParamState{ - Values: Params{EventId: 10}, - Has: PresenceMap{}, - }, - }, - }, - { - dataset: "dataset001_events/", - description: `select statement | selectors`, - output: `SELECT t.ID, t.Price FROM (SELECT * FROM EVENTS ORDER BY Price LIMIT 100 OFFSET 10) AS t WHERE price > 10`, - view: &view.View{ - Columns: []*view.Column{ - { - Name: "ID", - DataType: "Int", - }, - { - Name: "Price", - DataType: "Float", - }, - }, - Name: "events", - From: "SELECT * FROM EVENTS", - Table: "Events", - Template: &view.Template{ - Output: state.NewSchema(reflect.TypeOf(Params{})), - InputParameters: []*state.Parameter{ - { - Name: "EventId", - In: &state.State{ - Kind: state.KindPath, - Name: "eventId", - }, - Output: &state.Output{ - DataType: "int", - }, - }, - }, - }, - }, - selector: &view.Statelet{ - OrderBy: "price", - Criteria: "price > 10", - Limit: 100, - Offset: 10, - InputParameters: view.ParamState{ - Values: Params{}, - Has: PresenceMap{}, - }, - }, - }, - } - - //for index, useCase := range useCases[len(useCases)-1:] { - for index, useCase := range useCases { - tests.LogHeader(fmt.Sprintf("Running testcase nr: %v | %v \n", index, useCase.description)) - resourcePath := path.Join(testLocation, "testdata", "datasets", useCase.dataset, "populate") - if initDb(t, path.Join(testLocation, "testdata", "db_config.yaml"), resourcePath, "db") { - return - } - - useCase.view.Connector = &view.Connector{ - Name: "db", - DSN: "./testdata/db/db.db", - Driver: "sqlite3", - } - - if !assert.Nil(t, useCase.view.init(context.TODO(), view.EmptyResource()), useCase.description) { - continue - } +func TestActualLimit_PrefersSelectorNoLimit(t *testing.T) { + aView := &view.View{Selector: &view.Config{Limit: 1}} + selector := &view.Statelet{} + selector.WarmupNoLimit = true + selector.Limit = 0 - builder := NewBuilder() - - useCase.selector.init(useCase.view) - matcher, err := builder.Build(useCase.view, useCase.selector, useCase.batchData, useCase.relation, nil, nil, nil) - - assert.Nil(t, err, useCase.description) - assertly.AssertValues(t, useCase.placeholders, matcher.Args, useCase.description) - assert.Equal(t, useCase.output, strings.TrimSpace(matcher.SQL), useCase.description) - } -} - -func initDb(t *testing.T, configPath, datasetPath, dataStore string) bool { - datasetPath = datasetPath + "_" + dataStore - if !dsunit.InitFromURL(t, configPath) { - return true - } - - initDataset := dsunit.NewDatasetResource(dataStore, datasetPath, "", "") - request := dsunit.NewPrepareRequest(initDataset) - if !dsunit.Prepare(t, request) { - return true - } - - return false + assert.Equal(t, 0, actualLimit(aView, selector)) } - - -*/ diff --git a/view/cache.go b/view/cache.go index 557c97f7c..83075319a 100644 --- a/view/cache.go +++ b/view/cache.go @@ -12,7 +12,9 @@ import ( "github.com/viant/sqlx/io/read/cache" "github.com/viant/sqlx/io/read/cache/aerospike" "github.com/viant/sqlx/io/read/cache/afs" + "github.com/viant/tagly/format" rdata "github.com/viant/toolbox/data" + "reflect" "strconv" "strings" "sync" @@ -52,6 +54,7 @@ type ( IndexColumn string IndexParameter string `json:",omitempty" yaml:",omitempty"` IndexMeta bool `json:",omitempty"` + Limit *int `json:",omitempty" yaml:",omitempty"` FieldNames []string `json:",omitempty" yaml:",omitempty"` Connector *Connector `json:",omitempty"` Cases []*CacheParameters @@ -68,7 +71,9 @@ type ( // ExcludeDefault keeps explicitly declared warmup cases from adding an extra nil/default selector. ExcludeDefault bool `json:",omitempty" yaml:",omitempty"` - _param *state.Parameter + _param *state.Parameter + _location *time.Location + _locationInit bool } CacheInput struct { @@ -89,6 +94,8 @@ const ( aerospikeType = "aerospike" ) +var warmupNow = time.Now + func (c Caches) Has(name string) bool { for _, candidate := range c { if candidate.Name == name { @@ -392,6 +399,7 @@ func (c *Cache) generateDatasetSelectorsErr(ctx context.Context, set *CacheParam func (c *Cache) getParamValues(ctx context.Context, paramValue *ParamValue) ([]interface{}, error) { result := make([]interface{}, len(paramValue.Values), len(paramValue.Values)+1) for i, value := range paramValue.Values { + value = resolveWarmupValue(value, paramValue._param, paramValue._location) marshal := fmt.Sprintf("%v", value) converted, _, err := converter.Convert(marshal, paramValue._param.Schema.Type(), false, paramValue._param.DateFormat) if err != nil { @@ -407,6 +415,65 @@ func (c *Cache) getParamValues(ctx context.Context, paramValue *ParamValue) ([]i return result, nil } +func resolveWarmupValue(value interface{}, param *state.Parameter, location *time.Location) interface{} { + raw, ok := value.(string) + if !ok { + return value + } + raw = strings.TrimSpace(raw) + if raw == "" || !strings.HasPrefix(raw, "@") { + return value + } + now := warmupReferenceTime(location) + switch strings.ToLower(raw) { + case "@today": + return formatWarmupDate(now, param) + case "@yesterday": + return formatWarmupDate(now.AddDate(0, 0, -1), param) + default: + return value + } +} + +func warmupReferenceTime(location *time.Location) time.Time { + now := warmupNow() + if location == nil { + return now.UTC() + } + return now.In(location) +} + +func warmupLocation(param *state.Parameter) (*time.Location, error) { + if param == nil || strings.TrimSpace(param.Tag) == "" { + return nil, nil + } + parsed, err := format.Parse(reflect.StructTag(param.Tag)) + if err != nil { + return nil, fmt.Errorf("invalid warmup format tag on parameter %s: %w", param.Name, err) + } + if parsed == nil || strings.TrimSpace(parsed.Timezone) == "" { + return nil, nil + } + switch timezone := strings.TrimSpace(parsed.Timezone); timezone { + case "UTC", "utc": + return time.UTC, nil + default: + location, loadErr := time.LoadLocation(timezone) + if loadErr != nil { + return nil, fmt.Errorf("invalid warmup timezone %q on parameter %s: %w", timezone, param.Name, loadErr) + } + return location, nil + } +} + +func formatWarmupDate(value time.Time, param *state.Parameter) string { + layout := "2006-01-02" + if param != nil && strings.TrimSpace(param.DateFormat) != "" { + layout = param.DateFormat + } + return value.Format(layout) +} + func (c *Cache) initWarmup(ctx context.Context, resource *Resource) error { if c.owner == nil || c.Warmup == nil { return nil @@ -447,16 +514,24 @@ func (c *Cache) initWarmup(ctx context.Context, resource *Resource) error { } func (c *Cache) ensureParam(paramValue *ParamValue) error { - if paramValue._param != nil { - return nil + param := paramValue._param + if param == nil { + var err error + param, err = c.owner.Template._parametersIndex.Lookup(paramValue.Name) + if err != nil { + return err + } + paramValue._param = param } - param, err := c.owner.Template._parametersIndex.Lookup(paramValue.Name) - if err != nil { - return err + if !paramValue._locationInit { + location, err := warmupLocation(param) + if err != nil { + return err + } + paramValue._location = location + paramValue._locationInit = true } - - paramValue._param = param return nil } @@ -545,6 +620,10 @@ func (c *Cache) NewInput(selector *Statelet) *CacheInput { func (c *Cache) newInput(selector *Statelet, set *CacheParameters) *CacheInput { fieldNames := c.fieldNamesFor(set) + if selector != nil && c.Warmup != nil && c.Warmup.Limit != nil { + selector.Limit = *c.Warmup.Limit + selector.WarmupNoLimit = *c.Warmup.Limit == 0 + } c.applyWarmupFieldNames(selector, fieldNames) return &CacheInput{ Selector: selector, diff --git a/view/cache_warmup_test.go b/view/cache_warmup_test.go new file mode 100644 index 000000000..91cf58b9f --- /dev/null +++ b/view/cache_warmup_test.go @@ -0,0 +1,127 @@ +package view + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/viant/datly/view/state" +) + +func TestResolveWarmupValue_RelativeDates(t *testing.T) { + prevNow := warmupNow + warmupNow = func() time.Time { + return time.Date(2026, time.June, 30, 23, 30, 0, 0, time.UTC) + } + defer func() { + warmupNow = prevNow + }() + + param := &state.Parameter{DateFormat: "2006-01-02"} + + today := resolveWarmupValue("@today", param, nil) + yesterday := resolveWarmupValue("@yesterday", param, nil) + + assert.Regexp(t, `^\d{4}-\d{2}-\d{2}$`, today) + assert.Regexp(t, `^\d{4}-\d{2}-\d{2}$`, yesterday) + assert.NotEqual(t, today, yesterday) +} + +func TestResolveWarmupValue_PreservesLiteralValues(t *testing.T) { + param := &state.Parameter{DateFormat: "2006-01-02"} + + assert.Equal(t, "today", resolveWarmupValue("today", param, nil)) + assert.Equal(t, "@unsupported", resolveWarmupValue("@unsupported", param, nil)) + assert.Equal(t, 7, resolveWarmupValue(7, param, nil)) +} + +func TestResolveWarmupValue_UsesParameterTimezone(t *testing.T) { + prevNow := warmupNow + warmupNow = func() time.Time { + return time.Date(2026, time.July, 1, 0, 30, 0, 0, time.UTC) + } + defer func() { + warmupNow = prevNow + }() + + param := &state.Parameter{ + DateFormat: "2006-01-02", + Tag: `format:"dateFormat=2006-01-02,tz=America/New_York"`, + } + + location, err := warmupLocation(param) + assert.NoError(t, err) + assert.NotNil(t, location) + assert.Equal(t, "2026-06-30", resolveWarmupValue("@today", param, location)) + assert.Equal(t, "2026-06-29", resolveWarmupValue("@yesterday", param, location)) +} + +func TestWarmupLocation_RejectsInvalidTimezone(t *testing.T) { + param := &state.Parameter{ + Name: "From", + Tag: `format:"dateFormat=2006-01-02,tz=America/NewYork"`, + } + + location, err := warmupLocation(param) + assert.Nil(t, location) + assert.Error(t, err) +} + +func TestWarmupLocation_RejectsInvalidFormatTag(t *testing.T) { + param := &state.Parameter{ + Name: "From", + Tag: `format:"bogus=value"`, + } + + location, err := warmupLocation(param) + assert.Nil(t, location) + assert.Error(t, err) +} + +func TestInitWarmup_RejectsInvalidTimezoneOnGeneratedOptionalParam(t *testing.T) { + param := state.NewParameter("From", state.NewQueryLocation("from")) + param.Tag = `format:"dateFormat=2006-01-02,tz=America/NewYork"` + aView := &View{ + Name: "events", + Columns: []*Column{ + {Name: "event_type_id"}, + }, + Template: NewTemplate("", WithTemplateParameters(param)), + Cache: &Cache{ + Warmup: &Warmup{IndexColumn: "event_type_id"}, + }, + } + aView.Template._parametersIndex = aView.Template.Parameters.Index() + aView.indexColumns() + aView.Cache.owner = aView + + err := aView.Cache.initWarmup(context.Background(), EmptyResource()) + + assert.Error(t, err) +} + +func TestInitWarmup_SetsTimezoneOnGeneratedOptionalParam(t *testing.T) { + param := state.NewParameter("From", state.NewQueryLocation("from")) + param.Tag = `format:"dateFormat=2006-01-02,tz=America/New_York"` + aView := &View{ + Name: "events", + Columns: []*Column{ + {Name: "event_type_id"}, + }, + Template: NewTemplate("", WithTemplateParameters(param)), + Cache: &Cache{ + Warmup: &Warmup{IndexColumn: "event_type_id"}, + }, + } + aView.Template._parametersIndex = aView.Template.Parameters.Index() + aView.indexColumns() + aView.Cache.owner = aView + + err := aView.Cache.initWarmup(context.Background(), EmptyResource()) + + assert.NoError(t, err) + if assert.Len(t, aView.Cache.Warmup.Cases, 1) && assert.Len(t, aView.Cache.Warmup.Cases[0].Set, 1) { + assert.NotNil(t, aView.Cache.Warmup.Cases[0].Set[0]._location) + } +} diff --git a/view/state.go b/view/state.go index 370b9d386..48761aa7e 100644 --- a/view/state.go +++ b/view/state.go @@ -22,10 +22,11 @@ type ( Template *structology.State state.QuerySelector QuerySettings - filtersMu sync.Mutex - initialized bool - _columnNames map[string]bool - result *cache.ParmetrizedQuery + filtersMu sync.Mutex + initialized bool + WarmupNoLimit bool + _columnNames map[string]bool + result *cache.ParmetrizedQuery predicate.Filters Ignore bool } @@ -166,6 +167,7 @@ func (s *Statelet) CloneForSummary() *Statelet { QuerySelector: s.QuerySelector, QuerySettings: s.QuerySettings, initialized: s.initialized, + WarmupNoLimit: s.WarmupNoLimit, result: s.result, Ignore: s.Ignore, } diff --git a/warmup/cache_test.go b/warmup/cache_test.go index 6eb6c3527..aba4b86c4 100644 --- a/warmup/cache_test.go +++ b/warmup/cache_test.go @@ -242,6 +242,106 @@ Views: assert.Contains(t, fieldQuery.SQL, "quantity") } +func TestGenerateCacheInput_AppliesWarmupLimitOverride(t *testing.T) { + resourcePath := path.Join(t.TempDir(), "resource.yaml") + require.NoError(t, os.WriteFile(resourcePath, []byte(` +CacheProviders: + - Name: aerospike + Location: ${view.Name} + Provider: 'aerospike://127.0.0.1:3000/test' + TimeToLiveMs: 3600000 + +Connectors: + - Name: db + Driver: sqlite3 + DSN: ":memory:" + +Views: + - Name: events + Connector: + Ref: db + Table: events + Columns: + - Name: event_type_id + DataType: int + - Name: quantity + DataType: int + Cache: + Ref: aerospike + Warmup: + IndexColumn: event_type_id + Limit: 25 + Selector: + Constraints: + Limit: true + Limit: 1 + Template: + Source: SELECT * FROM EVENTS +`), 0644)) + + resource, err := view.NewResourceFromURL(context.Background(), resourcePath, nil, nil) + require.NoError(t, err) + require.NotEmpty(t, resource.Views) + aView := resource.Views[0] + + input, err := aView.Cache.GenerateCacheInput(context.Background()) + require.NoError(t, err) + require.Len(t, input, 1) + require.NotNil(t, input[0].Selector) + assert.Equal(t, 25, input[0].Selector.Limit) + assert.False(t, input[0].Selector.WarmupNoLimit) +} + +func TestGenerateCacheInput_ZeroWarmupLimitSetsNoLimit(t *testing.T) { + resourcePath := path.Join(t.TempDir(), "resource.yaml") + require.NoError(t, os.WriteFile(resourcePath, []byte(` +CacheProviders: + - Name: aerospike + Location: ${view.Name} + Provider: 'aerospike://127.0.0.1:3000/test' + TimeToLiveMs: 3600000 + +Connectors: + - Name: db + Driver: sqlite3 + DSN: ":memory:" + +Views: + - Name: events + Connector: + Ref: db + Table: events + Columns: + - Name: event_type_id + DataType: int + - Name: quantity + DataType: int + Cache: + Ref: aerospike + Warmup: + IndexColumn: event_type_id + Limit: 0 + Selector: + Constraints: + Limit: true + Limit: 1 + Template: + Source: SELECT * FROM EVENTS +`), 0644)) + + resource, err := view.NewResourceFromURL(context.Background(), resourcePath, nil, nil) + require.NoError(t, err) + require.NotEmpty(t, resource.Views) + aView := resource.Views[0] + + input, err := aView.Cache.GenerateCacheInput(context.Background()) + require.NoError(t, err) + require.Len(t, input, 1) + require.NotNil(t, input[0].Selector) + assert.Equal(t, 0, input[0].Selector.Limit) + assert.True(t, input[0].Selector.WarmupNoLimit) +} + func checkIfCached(t *testing.T, cache *view.Cache, ctx context.Context, testCase struct { description string URL string