diff --git a/go.mod b/go.mod index ae00e07c..74736b69 100644 --- a/go.mod +++ b/go.mod @@ -28,7 +28,7 @@ require ( github.com/viant/parsly v0.3.3 github.com/viant/pgo v0.11.0 github.com/viant/scy v0.33.1 - github.com/viant/sqlx v0.23.1-0.20260721202550-583cf232e734 + github.com/viant/sqlx v0.23.1-0.20260729142839-24da78934871 github.com/viant/structql v0.5.4 github.com/viant/toolbox v0.37.0 github.com/viant/velty v0.4.1-0.20260408224432-5a1c31e1bd87 diff --git a/go.sum b/go.sum index 109d15f6..029eec27 100644 --- a/go.sum +++ b/go.sum @@ -1198,6 +1198,8 @@ github.com/viant/sqlparser v0.12.1-0.20260409013525-147f8fc299b7 h1:2FdVturjHBSQ github.com/viant/sqlparser v0.12.1-0.20260409013525-147f8fc299b7/go.mod h1:2QRGiGZYk2/pjhORGG1zLVQ9JO+bXFhqIVi31mkCRPg= github.com/viant/sqlx v0.23.1-0.20260721202550-583cf232e734 h1:vZF9F8r3lUSfdRBMZyWje0eabeI0Q5sMwbd0QF3pq8c= github.com/viant/sqlx v0.23.1-0.20260721202550-583cf232e734/go.mod h1:dizufL+nTNqDCpivUnE2HqtddTp2TdA6WFghGfZo11c= +github.com/viant/sqlx v0.23.1-0.20260729142839-24da78934871 h1:9RqxSYtQfUGiMoT9YNpBfPZVjtorUrJ3uQISvS43EKI= +github.com/viant/sqlx v0.23.1-0.20260729142839-24da78934871/go.mod h1:dizufL+nTNqDCpivUnE2HqtddTp2TdA6WFghGfZo11c= github.com/viant/structology v0.9.0 h1:ibR/XmdQ3+/4XW3JK+pXRqugSnxJOm2bmIvbQ0hqztY= github.com/viant/structology v0.9.0/go.mod h1:AAFeViwniqua61sTKdOz/zlbLpN5vE4OVhDoiZJaMgA= github.com/viant/structql v0.5.4 h1:bMdcOpzU8UMoe5OBcyJVRxLAndvU1oj3ysvPUgBckCI= diff --git a/repository/locator/component/component.go b/repository/locator/component/component.go index 86c9f38f..0e37a12e 100644 --- a/repository/locator/component/component.go +++ b/repository/locator/component/component.go @@ -7,10 +7,12 @@ import ( "net/http" "net/url" "reflect" + "strings" "github.com/viant/datly/repository/contract" "github.com/viant/datly/service/executor/uow" "github.com/viant/datly/shared" + "github.com/viant/datly/view" "github.com/viant/datly/view/state" "github.com/viant/datly/view/state/kind" "github.com/viant/datly/view/state/kind/locator" @@ -48,11 +50,12 @@ func (l *componentLocator) Value(ctx context.Context, _ reflect.Type, name strin if err != nil { return nil, false, err } + request = sanitizeSelectorRequest(request) form := l.form value, err := l.dispatch.Dispatch(ctx, &contract.Path{Method: method, URI: URI}, contract.WithRequest(request), contract.WithConstants(l.constants), contract.WithPath(l.path), - contract.WithQuery(l.query), + contract.WithQuery(sanitizeSelectorQuery(l.query)), contract.WithForm(form), contract.WithLogger(l.logger), contract.WithHeader(l.header), @@ -61,6 +64,61 @@ func (l *componentLocator) Value(ctx context.Context, _ reflect.Type, name strin return value, err == nil, err } +func sanitizeSelectorQuery(query url.Values) url.Values { + sanitized, _ := sanitizeSelectorQueryWithRemoval(query) + return sanitized +} + +func sanitizeSelectorQueryWithRemoval(query url.Values) (url.Values, bool) { + if len(query) == 0 { + return query, false + } + removed := false + result := make(url.Values, len(query)) + for key, values := range query { + if isSelectorQueryKey(key) { + removed = true + continue + } + result[key] = append([]string(nil), values...) + } + if !removed { + return query, false + } + return result, true +} + +func sanitizeSelectorRequest(request *http.Request) *http.Request { + if request == nil || request.URL == nil || request.URL.RawQuery == "" { + return request + } + sanitized, removed := sanitizeSelectorQueryWithRemoval(request.URL.Query()) + if !removed { + return request + } + cloned := request.Clone(request.Context()) + cloned.URL = cloneURL(request.URL) + cloned.URL.RawQuery = sanitized.Encode() + return cloned +} + +func cloneURL(src *url.URL) *url.URL { + if src == nil { + return nil + } + cloned := *src + return &cloned +} + +func isSelectorQueryKey(name string) bool { + switch strings.ToLower(strings.TrimSpace(name)) { + case view.FieldsQuery, view.OrderByQuery, view.LimitQuery, view.OffsetQuery, view.PageQuery, view.CriteriaQuery: + return true + default: + return false + } +} + func updateErrWithResponseStatus(err error, response interface{}) error { var statusErr error responseStatus, ok := tryExtractResponseStatus(response) diff --git a/repository/locator/component/component_uow_test.go b/repository/locator/component/component_uow_test.go index f92e17cd..3a7c419d 100644 --- a/repository/locator/component/component_uow_test.go +++ b/repository/locator/component/component_uow_test.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "net/http" + "net/url" "reflect" "testing" @@ -39,6 +40,18 @@ func (d *componentScopeDispatcher) Dispatch(ctx context.Context, path *contract. return struct{}{}, nil } +type componentQueryDispatcher struct { + query url.Values + request *http.Request +} + +func (d *componentQueryDispatcher) Dispatch(_ context.Context, _ *contract.Path, opts ...contract.Option) (interface{}, error) { + options := contract.NewOptions(opts...) + d.query = options.Query + d.request = options.Request + return struct{}{}, nil +} + func TestComponentLocatorCreatesOrderedBindingFrames(t *testing.T) { db, _ := sql.Open("sqlite3", ":memory:") defer db.Close() @@ -79,6 +92,120 @@ func TestComponentLocatorCreatesOrderedBindingFrames(t *testing.T) { } } +func TestComponentLocatorDropsSelectorQueryParamsForChildDispatch(t *testing.T) { + dispatcher := &componentQueryDispatcher{} + request, _ := http.NewRequest(http.MethodGet, "/?_fields=AudienceId&_orderby=AudienceId&_limit=10&_offset=5&_page=2&_criteria=AudienceId+%3D+1&criteria=business+criteria&audience_id=123&order_id=456&from=2026-07-01&to=2026-07-02", nil) + query := url.Values{ + "_fields": {"AudienceId"}, + "_orderby": {"AudienceId"}, + "_limit": {"10"}, + "_offset": {"5"}, + "_page": {"2"}, + "_criteria": {"AudienceId = 1"}, + "criteria": {"business criteria"}, + "audience_id": {"123"}, + "order_id": {"456"}, + "from": {"2026-07-01"}, + "to": {"2026-07-02"}, + } + componentLocator := &componentLocator{ + dispatch: dispatcher, + query: query, + getRequest: func() (*http.Request, error) { + return request, nil + }, + } + + _, found, err := componentLocator.Value(context.Background(), reflect.TypeOf(""), "GET:/child") + if err != nil || !found { + t.Fatalf("Value() found=%v err=%v", found, err) + } + + for _, key := range []string{"_fields", "_orderby", "_limit", "_offset", "_page", "_criteria"} { + if _, ok := dispatcher.query[key]; ok { + t.Fatalf("selector query key %q was forwarded: %v", key, dispatcher.query) + } + } + for key, want := range map[string]string{ + "criteria": "business criteria", + "audience_id": "123", + "order_id": "456", + "from": "2026-07-01", + "to": "2026-07-02", + } { + if got := dispatcher.query.Get(key); got != want { + t.Fatalf("query[%s]=%q want %q; query=%v", key, got, want, dispatcher.query) + } + } + if dispatcher.request == nil || dispatcher.request.URL == nil { + t.Fatal("expected forwarded request") + } + requestQuery := dispatcher.request.URL.Query() + for _, key := range []string{"_fields", "_orderby", "_limit", "_offset", "_page", "_criteria"} { + if _, ok := requestQuery[key]; ok { + t.Fatalf("selector query key %q was forwarded on request URL: %s", key, dispatcher.request.URL.RawQuery) + } + } + for key, want := range map[string]string{ + "criteria": "business criteria", + "audience_id": "123", + "order_id": "456", + "from": "2026-07-01", + "to": "2026-07-02", + } { + if got := requestQuery.Get(key); got != want { + t.Fatalf("request query[%s]=%q want %q; raw=%s", key, got, want, dispatcher.request.URL.RawQuery) + } + } + if request.URL.Query().Get("_fields") != "AudienceId" { + t.Fatal("original parent request was mutated") + } +} + +func TestSanitizeSelectorQueryClonesForwardedValues(t *testing.T) { + query := url.Values{"_fields": {"AudienceId"}, "order_id": {"456"}} + sanitized := sanitizeSelectorQuery(query) + + query.Set("order_id", "mutated") + + if got, want := sanitized.Get("order_id"), "456"; got != want { + t.Fatalf("sanitized query was not cloned, got %q want %q", got, want) + } +} + +func TestSanitizeSelectorRequestClonesForwardedRequest(t *testing.T) { + request, _ := http.NewRequest(http.MethodGet, "/?_fields=AudienceId&order_id=456", nil) + + sanitized := sanitizeSelectorRequest(request) + + if sanitized == request { + t.Fatal("expected sanitized request clone") + } + if got := sanitized.URL.Query().Get("_fields"); got != "" { + t.Fatalf("sanitized request still has _fields=%q", got) + } + if got, want := sanitized.URL.Query().Get("order_id"), "456"; got != want { + t.Fatalf("sanitized request order_id=%q want %q", got, want) + } + if got, want := request.URL.Query().Get("_fields"), "AudienceId"; got != want { + t.Fatalf("original request was mutated, _fields=%q want %q", got, want) + } +} + +func TestSanitizeSelectorRequestDoesNotReencodeWhenNoSelectorParams(t *testing.T) { + request, _ := http.NewRequest(http.MethodGet, "/?b=two%20words&a=1", nil) + originalRawQuery := request.URL.RawQuery + + sanitized := sanitizeSelectorRequest(request) + + if sanitized != request { + t.Fatal("expected original request when no selector params are present") + } + if sanitized.URL.RawQuery != originalRawQuery { + t.Fatalf("raw query changed, got %q want %q", sanitized.URL.RawQuery, originalRawQuery) + } +} + func TestComponentLocatorRequiresInvocationDispatcher(t *testing.T) { if _, err := newComponentLocator(locator.WithConstants(nil)); err == nil { t.Fatal("expected missing dispatcher error") diff --git a/service.go b/service.go index 98d5ac02..27cf7a25 100644 --- a/service.go +++ b/service.go @@ -186,6 +186,22 @@ func WithOutput(output interface{}) OperateOption { } } +func ContextWithOutputProjection(ctx context.Context, output interface{}) context.Context { + return session.ContextWithOutputProjection(ctx, output) +} + +func ContextWithViewOutputProjection(ctx context.Context, viewName string, output interface{}) context.Context { + return session.ContextWithViewOutputProjection(ctx, viewName, output) +} + +func ContextWithOutputFields(ctx context.Context, fields ...string) context.Context { + return session.ContextWithOutputFields(ctx, fields...) +} + +func ContextWithViewOutputFields(ctx context.Context, viewName string, fields ...string) context.Context { + return session.ContextWithViewOutputFields(ctx, viewName, fields...) +} + func WithSession(session *session.Session) OperateOption { return func(o *operateOptions) { o.session = session diff --git a/service/reader/handler/handler.go b/service/reader/handler/handler.go index 98a3e47d..3f568231 100644 --- a/service/reader/handler/handler.go +++ b/service/reader/handler/handler.go @@ -118,6 +118,9 @@ func (h *Handler) readData(ctx context.Context, aView *view.View, aState *sessio return err } } + if err = aState.ApplyOutputProjection(ctx, aView); err != nil { + return err + } if err = aState.Populate(ctx); err != nil { return err } diff --git a/service/reader/service.go b/service/reader/service.go index f635064c..70e7fb95 100644 --- a/service/reader/service.go +++ b/service/reader/service.go @@ -630,8 +630,83 @@ func (s *Service) warmupMatcher(ctx context.Context, aView *view.View, statelet } cloned := *statelet cloned.Template = clonedTemplate + ok, err := applyWarmupIdentityProjection(aView, &cloned) + if err != nil { + return nil, err + } + if !ok { + return nil, nil + } + + matcher, err := s.sqlBuilder.CacheSQLWithOptions(ctx, aView, &cloned, nil, nil, parent) + if err != nil || matcher == nil { + return matcher, err + } + if err = applyRequestedFields(aView, statelet, matcher); err != nil { + fmt.Printf("[INFO] datly warmup projection metadata error view=%s fields=%v error=%v\n", aView.Name, requestedFieldNames(statelet), err) + return nil, nil + } + return matcher, nil +} - return s.sqlBuilder.CacheSQLWithOptions(ctx, aView, &cloned, nil, nil, parent) +func applyWarmupIdentityProjection(aView *view.View, statelet *view.Statelet) (bool, error) { + if aView == nil || aView.Cache == nil || aView.Cache.Warmup == nil || statelet == nil { + return true, nil + } + fieldNames, ok := aView.Cache.WarmupFieldNamesForSelector(statelet) + if !ok { + return false, nil + } + if len(fieldNames) == 0 { + statelet.SetColumns(nil) + statelet.Fields = nil + return true, nil + } + columns, err := view.ProjectionColumnsForNames(aView, fieldNames) + if err != nil { + return false, err + } + fields := make([]string, 0, len(columns)) + for _, columnName := range columns { + column, ok := aView.ColumnByName(columnName) + if !ok { + return false, fmt.Errorf("failed to map warmup identity column %s to view %s column", columnName, aView.Name) + } + fieldName := column.FieldName() + if fieldName == "" { + fieldName = column.Name + } + fields = append(fields, fieldName) + } + statelet.SetColumns(columns) + statelet.Fields = fields + return true, nil +} + +func applyRequestedFields(aView *view.View, statelet *view.Statelet, matcher *cache.ParmetrizedQuery) error { + if aView == nil || statelet == nil || matcher == nil { + return nil + } + names := statelet.Columns + if len(names) == 0 { + names = statelet.Fields + } + fields, err := view.ProjectionFieldsForNames(aView, names) + if err != nil { + return err + } + matcher.RequestedFields = view.SQLXProjectionFields(fields) + return nil +} + +func requestedFieldNames(statelet *view.Statelet) []string { + if statelet == nil { + return nil + } + if len(statelet.Columns) != 0 { + return statelet.Columns + } + return statelet.Fields } func warmupIndexParameter(aView *view.View) *state.Parameter { diff --git a/service/reader/service_warmup_test.go b/service/reader/service_warmup_test.go index 53ca54d0..ea4aae56 100644 --- a/service/reader/service_warmup_test.go +++ b/service/reader/service_warmup_test.go @@ -8,6 +8,7 @@ import ( "github.com/stretchr/testify/require" "github.com/viant/datly/view" "github.com/viant/datly/view/state" + "github.com/viant/sqlx/io/read/cache" "github.com/viant/structology" ) @@ -205,3 +206,271 @@ func TestWarmupIndexParameterDoesNotMatchUnrelatedParameter(t *testing.T) { require.Nil(t, parameter) } + +func TestApplyRequestedFieldsPopulatesMatcherProjection(t *testing.T) { + aView := view.NewView("events", "events", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + {Name: "order_id", DataType: "int", Tag: `json:"orderId"`, Groupable: true}, + {Name: "bids", DataType: "int", Aggregate: true}, + {Name: "impressions", DataType: "int", Aggregate: true}, + }), + view.WithGroupable(true), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + statelet := view.NewStatelet() + statelet.SetColumns([]string{"impressions", "order_id"}) + matcher := &cache.ParmetrizedQuery{} + + require.NoError(t, applyRequestedFields(aView, statelet, matcher)) + + require.Len(t, matcher.RequestedFields, 2) + require.Equal(t, "impressions", matcher.RequestedFields[0].Name) + require.Equal(t, "impressions", matcher.RequestedFields[0].MeasureKey) + require.Equal(t, "order_id", matcher.RequestedFields[1].Name) + require.Equal(t, "order_id", matcher.RequestedFields[1].DimensionKey) +} + +func TestApplyRequestedFieldsIgnoresUnmappedProjection(t *testing.T) { + aView := view.NewView("events", "events", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + {Name: "order_id", DataType: "int"}, + }), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + statelet := view.NewStatelet() + statelet.SetColumns([]string{"missing"}) + matcher := &cache.ParmetrizedQuery{} + + require.Error(t, applyRequestedFields(aView, statelet, matcher)) + + require.Empty(t, matcher.RequestedFields) +} + +func TestApplyRequestedFieldsUsesFullProjectionWhenRequestHasNoProjection(t *testing.T) { + aView := view.NewView("events", "events", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + {Name: "audience_id", DataType: "int", Groupable: true}, + {Name: "spend", DataType: "float", Aggregate: true}, + }), + view.WithGroupable(true), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + matcher := &cache.ParmetrizedQuery{} + + require.NoError(t, applyRequestedFields(aView, view.NewStatelet(), matcher)) + + require.Len(t, matcher.RequestedFields, 2) + require.Equal(t, "audience_id", matcher.RequestedFields[0].DimensionKey) + require.Equal(t, "spend", matcher.RequestedFields[1].MeasureKey) +} + +func TestApplyWarmupIdentityProjectionUsesWarmupFieldNames(t *testing.T) { + aView := view.NewView("events", "events", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + {Name: "audience_id", DataType: "int", Tag: `json:"audienceId"`, Groupable: true}, + {Name: "bids", DataType: "int", Aggregate: true}, + {Name: "spend", DataType: "float", Aggregate: true}, + {Name: "period_ecpm", DataType: "float", Aggregate: true}, + }), + view.WithGroupable(true), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + aView.Cache = &view.Cache{ + Warmup: &view.Warmup{ + FieldNames: []string{"audience_id", "bids", "spend", "period_ecpm"}, + }, + } + statelet := view.NewStatelet() + statelet.SetColumns([]string{"audience_id", "spend", "period_ecpm"}) + statelet.Fields = []string{"AudienceId", "Spend", "PeriodEcpm"} + + ok, err := applyWarmupIdentityProjection(aView, statelet) + require.NoError(t, err) + require.True(t, ok) + + require.Equal(t, []string{"audience_id", "bids", "spend", "period_ecpm"}, statelet.Columns) + require.Equal(t, []string{"audience_id", "bids", "spend", "period_ecpm"}, statelet.Fields) +} + +func TestApplyWarmupIdentityProjectionUsesMatchingCaseFieldNames(t *testing.T) { + periodParam := state.NewParameter("Period", state.NewQueryLocation("period"), state.WithParameterType(reflect.TypeOf(""))) + aView := view.NewView("events", "events", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + {Name: "audience_id", DataType: "int", Groupable: true}, + {Name: "bids", DataType: "int", Aggregate: true}, + {Name: "spend", DataType: "float", Aggregate: true}, + {Name: "period_ecpm", DataType: "float", Aggregate: true}, + }), + view.WithTemplate(view.NewTemplate("", view.WithTemplateParameters(periodParam))), + view.WithGroupable(true), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + aView.Cache = &view.Cache{ + Warmup: &view.Warmup{ + FieldNames: []string{"audience_id", "bids"}, + Cases: []*view.CacheParameters{ + { + Set: []*view.ParamValue{ + {Name: "Period", Values: []interface{}{"today"}}, + }, + FieldNames: []string{"audience_id", "spend", "period_ecpm"}, + }, + }, + }, + } + statelet := view.NewStatelet() + statelet.Init(aView) + require.NoError(t, periodParam.Set(statelet.Template, "today")) + + ok, err := applyWarmupIdentityProjection(aView, statelet) + require.NoError(t, err) + require.True(t, ok) + + require.Equal(t, []string{"audience_id", "spend", "period_ecpm"}, statelet.Columns) +} + +func TestApplyWarmupIdentityProjectionMatchesDefaultOptionalCase(t *testing.T) { + periodParam := state.NewParameter("Period", state.NewQueryLocation("period"), state.WithParameterType(reflect.TypeOf(""))) + aView := view.NewView("events", "events", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + {Name: "audience_id", DataType: "int", Groupable: true}, + {Name: "bids", DataType: "int", Aggregate: true}, + {Name: "spend", DataType: "float", Aggregate: true}, + }), + view.WithTemplate(view.NewTemplate("", view.WithTemplateParameters(periodParam))), + view.WithGroupable(true), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + aView.Cache = &view.Cache{ + Warmup: &view.Warmup{ + FieldNames: []string{"audience_id", "bids"}, + Cases: []*view.CacheParameters{ + { + Set: []*view.ParamValue{ + {Name: "Period"}, + }, + FieldNames: []string{"audience_id", "spend"}, + }, + }, + }, + } + statelet := view.NewStatelet() + statelet.Init(aView) + + ok, err := applyWarmupIdentityProjection(aView, statelet) + require.NoError(t, err) + require.True(t, ok) + + require.Equal(t, []string{"audience_id", "spend"}, statelet.Columns) +} + +func TestApplyWarmupIdentityProjectionClearsProjectionForAmbiguousCaseFieldNames(t *testing.T) { + periodParam := state.NewParameter("Period", state.NewQueryLocation("period"), state.WithParameterType(reflect.TypeOf(""))) + aView := view.NewView("events", "events", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + {Name: "audience_id", DataType: "int", Groupable: true}, + {Name: "bids", DataType: "int", Aggregate: true}, + {Name: "spend", DataType: "float", Aggregate: true}, + }), + view.WithTemplate(view.NewTemplate("", view.WithTemplateParameters(periodParam))), + view.WithGroupable(true), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + aView.Cache = &view.Cache{ + Warmup: &view.Warmup{ + Cases: []*view.CacheParameters{ + { + Set: []*view.ParamValue{ + {Name: "Period"}, + }, + FieldNames: []string{"audience_id", "bids"}, + }, + { + Set: []*view.ParamValue{ + {Name: "Period"}, + }, + FieldNames: []string{"audience_id", "spend"}, + }, + }, + }, + } + statelet := view.NewStatelet() + statelet.Init(aView) + statelet.SetColumns([]string{"audience_id", "spend"}) + + ok, err := applyWarmupIdentityProjection(aView, statelet) + require.NoError(t, err) + require.False(t, ok) + + require.Equal(t, []string{"audience_id", "spend"}, statelet.Columns) + require.Empty(t, statelet.Fields) +} + +func TestApplyWarmupIdentityProjectionAllowsEquivalentCaseFieldAliases(t *testing.T) { + periodParam := state.NewParameter("Period", state.NewQueryLocation("period"), state.WithParameterType(reflect.TypeOf(""))) + aView := view.NewView("events", "events", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + {Name: "audience_id", DataType: "int", Tag: `json:"audienceId"`, Groupable: true}, + {Name: "spend", DataType: "float", Aggregate: true}, + }), + view.WithTemplate(view.NewTemplate("", view.WithTemplateParameters(periodParam))), + view.WithGroupable(true), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + aView.Cache = &view.Cache{ + Warmup: &view.Warmup{ + Cases: []*view.CacheParameters{ + { + Set: []*view.ParamValue{ + {Name: "Period"}, + }, + FieldNames: []string{"AudienceId", "Spend"}, + }, + { + Set: []*view.ParamValue{ + {Name: "Period"}, + }, + FieldNames: []string{"audience_id", "spend"}, + }, + }, + }, + } + statelet := view.NewStatelet() + statelet.Init(aView) + + ok, err := applyWarmupIdentityProjection(aView, statelet) + require.NoError(t, err) + require.True(t, ok) + + require.Equal(t, []string{"audience_id", "spend"}, statelet.Columns) +} + +func TestApplyWarmupIdentityProjectionClearsProjectionForFullWarmup(t *testing.T) { + aView := view.NewView("events", "events", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + {Name: "audience_id", DataType: "int"}, + {Name: "spend", DataType: "float"}, + }), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + aView.Cache = &view.Cache{Warmup: &view.Warmup{}} + statelet := view.NewStatelet() + statelet.SetColumns([]string{"audience_id", "spend"}) + statelet.Fields = []string{"AudienceId", "Spend"} + + ok, err := applyWarmupIdentityProjection(aView, statelet) + require.NoError(t, err) + require.True(t, ok) + + require.Empty(t, statelet.Columns) + require.Empty(t, statelet.Fields) +} diff --git a/service/reader/sql_groupable_test.go b/service/reader/sql_groupable_test.go index 16e652c3..1dcb8507 100644 --- a/service/reader/sql_groupable_test.go +++ b/service/reader/sql_groupable_test.go @@ -193,6 +193,59 @@ func TestBuilder_rewriteGroupBy(t *testing.T) { }, expected: "(SELECT (99 * SUM(avails)) + MOD(SUM(avails), 99) AS avails FROM audience_event_v1 v)", }, + { + description: "rewrite site cube shared diagnostic projection", + sql: "(SELECT ao.event_date, ao.advertiser_date, ao.agency_id, ao.advertiser_id, ao.campaign_id, " + + "ao.ad_order_id, ao.audience_id, ao.creative_id, ao.deal_id, ao.publisher_id, ao.channel_id, " + + "ao.country, ao.site_type, ao.site_id, SUM(ao.bids) AS bids, SUM(ao.impressions) AS impressions, " + + "SUM(ao.clicks) AS clicks, SUM(ao.conversions) AS conversions, SUM(ao.view_conversions) AS view_conversions, " + + "SUM(ao.total_spend) AS total_spend, SUM(ao.v_start) AS v_start, SUM(ao.v_100) AS v_100 " + + "FROM fact_perf_site_daily_v ao GROUP BY ao.event_date, ao.advertiser_date, ao.agency_id, ao.advertiser_id, " + + "ao.campaign_id, ao.ad_order_id, ao.audience_id, ao.creative_id, ao.deal_id, ao.external_deal_id, " + + "ao.publisher_id, ao.channel_id, ao.country, ao.site_type, ao.is_pg, ao.media_execution_id, ao.site_id)", + allColumns: []*view.Column{ + {Name: "event_date", Groupable: true}, + {Name: "advertiser_date", Groupable: true}, + {Name: "agency_id", Groupable: true}, + {Name: "advertiser_id", Groupable: true}, + {Name: "campaign_id", Groupable: true}, + {Name: "ad_order_id", Groupable: true}, + {Name: "audience_id", Groupable: true}, + {Name: "creative_id", Groupable: true}, + {Name: "deal_id", Groupable: true}, + {Name: "publisher_id", Groupable: true}, + {Name: "channel_id", Groupable: true}, + {Name: "country", Groupable: true}, + {Name: "site_type", Groupable: true}, + {Name: "site_id", Groupable: true}, + {Name: "bids"}, + {Name: "impressions"}, + {Name: "clicks"}, + {Name: "conversions"}, + {Name: "view_conversions"}, + {Name: "total_spend"}, + {Name: "v_start"}, + {Name: "v_100"}, + }, + projected: []*view.Column{ + {Name: "event_date", Groupable: true}, + {Name: "advertiser_date", Groupable: true}, + {Name: "site_id", Groupable: true}, + {Name: "site_type", Groupable: true}, + {Name: "bids"}, + {Name: "impressions"}, + {Name: "clicks"}, + {Name: "conversions"}, + {Name: "view_conversions"}, + {Name: "total_spend"}, + {Name: "v_start"}, + {Name: "v_100"}, + }, + expected: "(SELECT ao.event_date, ao.advertiser_date, ao.site_id, ao.site_type, SUM(ao.bids) AS bids, SUM(ao.impressions) AS impressions, " + + "SUM(ao.clicks) AS clicks, SUM(ao.conversions) AS conversions, SUM(ao.view_conversions) AS view_conversions, " + + "SUM(ao.total_spend) AS total_spend, SUM(ao.v_start) AS v_start, SUM(ao.v_100) AS v_100 " + + "FROM fact_perf_site_daily_v ao GROUP BY 1, 2, 3, 4)", + }, { description: "rewrite grouped aggregates matches reordered forecasting measures by alias not metadata order", sql: "(SELECT IFNULL(STRING_AGG(DISTINCT IAB[SAFE_OFFSET(0)], ', ' LIMIT 20), '') AS iab_cats, " + @@ -221,11 +274,11 @@ func TestBuilder_rewriteGroupBy(t *testing.T) { {Name: "hh_uniqs"}, {Name: "device_uniqs"}, }, - expected: "(SELECT (99 * SUM(avails)) + MOD(SUM(avails), 99) AS avails, " + + expected: "(SELECT AVG(v.clearing_price) AS min_clearing_price, " + + "MAX(v.clearing_price) AS max_clearing_price, " + + "(99 * SUM(avails)) + MOD(SUM(avails), 99) AS avails, " + "(9100 * APPROX_COUNT_DISTINCT(IF(avails = 100, COALESCE(alias_ip, IF(hhip_flag = 1, ip, NULL)), NULL))) AS hh_uniqs, " + - "(9100 * APPROX_COUNT_DISTINCT(uid)) AS device_uniqs, " + - "AVG(v.clearing_price) AS min_clearing_price, " + - "MAX(v.clearing_price) AS max_clearing_price " + + "(9100 * APPROX_COUNT_DISTINCT(uid)) AS device_uniqs " + "FROM audience_event_v1 v)", }, { diff --git a/service/session/option.go b/service/session/option.go index 40ee8c6c..48e85b6d 100644 --- a/service/session/option.go +++ b/service/session/option.go @@ -37,6 +37,8 @@ type ( preseedCache bool cacheDisabled bool sqlTx *sql.Tx + viewProjections map[string][]string + outputProjection *OutputProjection } Option func(o *Options) @@ -138,6 +140,38 @@ func WithLocatorOptions(options ...locator.Option) Option { } } +func WithViewProjectionColumns(viewName string, columns []string) Option { + return func(s *Options) { + if len(columns) == 0 { + return + } + if s.viewProjections == nil { + s.viewProjections = map[string][]string{} + } + s.viewProjections[viewName] = append([]string(nil), columns...) + } +} + +func WithOutputProjection(output interface{}) Option { + return WithViewOutputProjection("", output) +} + +func WithViewOutputProjection(viewName string, output interface{}) Option { + return func(s *Options) { + s.outputProjection = &OutputProjection{View: viewName, Output: output} + } +} + +func WithOutputFields(fields ...string) Option { + return WithViewOutputFields("", fields...) +} + +func WithViewOutputFields(viewName string, fields ...string) Option { + return func(s *Options) { + s.outputProjection = &OutputProjection{View: viewName, Fields: append([]string(nil), fields...), FieldsHint: true} + } +} + func WithStateResource(resource state.Resource) Option { return func(s *Options) { s.resource = resource diff --git a/service/session/projection.go b/service/session/projection.go new file mode 100644 index 00000000..a43e01ca --- /dev/null +++ b/service/session/projection.go @@ -0,0 +1,104 @@ +package session + +import ( + "context" + "fmt" +) + +type outputProjectionKey struct{} + +type OutputProjection struct { + View string + Output interface{} + Fields []string + FieldsHint bool +} + +func ContextWithOutputProjection(ctx context.Context, output interface{}) context.Context { + return ContextWithViewOutputProjection(ctx, "", output) +} + +func ContextWithViewOutputProjection(ctx context.Context, viewName string, output interface{}) context.Context { + return context.WithValue(ctx, outputProjectionKey{}, &OutputProjection{View: viewName, Output: output}) +} + +func ContextWithOutputFields(ctx context.Context, fields ...string) context.Context { + return ContextWithViewOutputFields(ctx, "", fields...) +} + +func ContextWithViewOutputFields(ctx context.Context, viewName string, fields ...string) context.Context { + return context.WithValue(ctx, outputProjectionKey{}, &OutputProjection{View: viewName, Fields: append([]string(nil), fields...), FieldsHint: true}) +} + +func OutputProjectionFromContext(ctx context.Context, viewName string) interface{} { + if ctx == nil { + return nil + } + value := ctx.Value(outputProjectionKey{}) + switch actual := value.(type) { + case *OutputProjection: + if actual == nil { + return nil + } + if actual.View != "" && normalizeViewProjectionName(actual.View) != normalizeViewProjectionName(viewName) { + return nil + } + return actual.Output + case OutputProjection: + if actual.View != "" && normalizeViewProjectionName(actual.View) != normalizeViewProjectionName(viewName) { + return nil + } + return actual.Output + } + return nil +} + +func OutputFieldsFromContext(ctx context.Context, viewName string) []string { + if ctx == nil { + return nil + } + value := ctx.Value(outputProjectionKey{}) + switch actual := value.(type) { + case *OutputProjection: + if actual == nil { + return nil + } + if actual.View != "" && normalizeViewProjectionName(actual.View) != normalizeViewProjectionName(viewName) { + return nil + } + return append([]string(nil), actual.Fields...) + case OutputProjection: + if actual.View != "" && normalizeViewProjectionName(actual.View) != normalizeViewProjectionName(viewName) { + return nil + } + return append([]string(nil), actual.Fields...) + } + return nil +} + +func OutputFieldsHintFromContext(ctx context.Context, viewName string) bool { + if ctx == nil { + return false + } + value := ctx.Value(outputProjectionKey{}) + switch actual := value.(type) { + case *OutputProjection: + if actual == nil { + return false + } + if actual.View != "" && normalizeViewProjectionName(actual.View) != normalizeViewProjectionName(viewName) { + return false + } + return actual.FieldsHint + case OutputProjection: + if actual.View != "" && normalizeViewProjectionName(actual.View) != normalizeViewProjectionName(viewName) { + return false + } + return actual.FieldsHint + } + return false +} + +func EmptyOutputFieldsError(viewName string) error { + return fmt.Errorf("output projection for view %s did not specify any field names", viewName) +} diff --git a/service/session/reader.go b/service/session/reader.go index 94a91d52..4574d0b8 100644 --- a/service/session/reader.go +++ b/service/session/reader.go @@ -18,6 +18,9 @@ func (s *Session) ReadInto(ctx context.Context, dest interface{}, aView *view.Vi } }() } + if err := s.ApplyOutputProjection(ctx, aView); err != nil { + return err + } if err := s.SetViewState(ctx, aView); err != nil { return err } diff --git a/service/session/selector.go b/service/session/selector.go index 78818d03..7183b411 100644 --- a/service/session/selector.go +++ b/service/session/selector.go @@ -100,6 +100,9 @@ func (s *Session) setQuerySelector(ctx context.Context, ns *view.NamespaceView, // but still validate against view selector constraints. if injected != nil { selector.QuerySelector = injected.QuerySelector + if len(injected.Columns) > 0 { + selector.SetColumns(injected.Columns) + } if err := s.applyInjectedQuerySelector(ns, selector, injected); err != nil { return err } diff --git a/service/session/state.go b/service/session/state.go index e5754242..c00b7428 100644 --- a/service/session/state.go +++ b/service/session/state.go @@ -177,9 +177,118 @@ func (s *Session) setViewState(ctx context.Context, aView *view.View) (err error return err } } + s.applyViewProjection(aView) return err } +func (s *Session) applyViewProjection(aView *view.View) { + if s == nil || aView == nil || len(s.viewProjections) == 0 { + return + } + columns, ok := s.viewProjections[aView.Name] + if !ok { + normalizedViewName := normalizeViewProjectionName(aView.Name) + for name, candidate := range s.viewProjections { + if normalizeViewProjectionName(name) == normalizedViewName { + columns = candidate + ok = true + break + } + } + } + if !ok || len(columns) == 0 { + return + } + statelet := s.state.Lookup(aView) + statelet.SetColumns(columns) +} + +func (s *Session) ApplyOutputProjection(ctx context.Context, aView *view.View) error { + if hasFieldsHint := s.outputFieldsHintForView(ctx, aView.Name); hasFieldsHint { + fields := s.outputFieldsForView(ctx, aView.Name) + if len(fields) == 0 { + return EmptyOutputFieldsError(aView.Name) + } + columns, err := view.ProjectionColumnsForNames(aView, fields) + if err != nil { + return err + } + s.applyProjectionColumns(aView, columns) + return nil + } + var output interface{} + if s.outputProjection != nil { + output = projectionOutputForView(*s.outputProjection, aView.Name) + } + if output == nil { + output = OutputProjectionFromContext(ctx, aView.Name) + } + if output == nil { + return nil + } + columns, err := view.ProjectionColumnsForOutput(aView, output) + if err != nil { + return err + } + if columns == nil { + return nil + } + s.applyProjectionColumns(aView, columns) + return nil +} + +func (s *Session) outputFieldsForView(ctx context.Context, viewName string) []string { + if s.outputProjection != nil && len(s.outputProjection.Fields) > 0 { + return projectionFieldsForView(*s.outputProjection, viewName) + } + return OutputFieldsFromContext(ctx, viewName) +} + +func (s *Session) outputFieldsHintForView(ctx context.Context, viewName string) bool { + if s.outputProjection != nil && s.outputProjection.FieldsHint { + return projectionFieldsHintForView(*s.outputProjection, viewName) + } + return OutputFieldsHintFromContext(ctx, viewName) +} + +func (s *Session) applyProjectionColumns(aView *view.View, columns []string) { + if len(columns) == 0 { + return + } + s.Apply(WithViewProjectionColumns(aView.Name, columns)) + statelet := s.state.Lookup(aView) + statelet.SetColumns(columns) +} + +func projectionOutputForView(projection OutputProjection, viewName string) interface{} { + if projection.View != "" && normalizeViewProjectionName(projection.View) != normalizeViewProjectionName(viewName) { + return nil + } + return projection.Output +} + +func projectionFieldsForView(projection OutputProjection, viewName string) []string { + if projection.View != "" && normalizeViewProjectionName(projection.View) != normalizeViewProjectionName(viewName) { + return nil + } + return append([]string(nil), projection.Fields...) +} + +func projectionFieldsHintForView(projection OutputProjection, viewName string) bool { + if projection.View != "" && normalizeViewProjectionName(projection.View) != normalizeViewProjectionName(viewName) { + return false + } + return projection.FieldsHint +} + +func normalizeViewProjectionName(name string) string { + name = strings.ToLower(strings.TrimSpace(name)) + name = strings.ReplaceAll(name, "_", "") + name = strings.ReplaceAll(name, "-", "") + name = strings.ReplaceAll(name, ".", "") + return name +} + func (s *Session) viewNamespace(aView *view.View) *view.NamespaceView { ns := s.namespacedView.ByName(aView.Name) if ns == nil { diff --git a/service_projection_test.go b/service_projection_test.go new file mode 100644 index 00000000..83ce0afc --- /dev/null +++ b/service_projection_test.go @@ -0,0 +1,412 @@ +package datly + +import ( + "context" + "reflect" + "testing" + + "github.com/stretchr/testify/require" + "github.com/viant/datly/repository" + "github.com/viant/datly/repository/contract" + "github.com/viant/datly/service/reader" + "github.com/viant/datly/service/session" + "github.com/viant/datly/view" + "github.com/viant/datly/view/state" +) + +type fullProjectionOutput struct { + AccountID int `json:"accountId" sqlx:"account_id"` + CampaignID int `json:"campaignId" sqlx:"campaign_id"` + Impressions int `json:"impressions"` + Spend float64 `json:"spend"` +} + +type alternateProjectionOutput struct { + Campaign int `json:"campaignId"` + Spend int `sqlx:"spend"` +} + +type sourceProjectionOutput struct { + AliasValue int `source:"src.alias_value"` +} + +type ignoredEmbeddedProjection struct { + AccountID int `json:"accountId"` +} + +func TestProjectionColumnsForOutput_BuildsProjectionFromDestinationType(t *testing.T) { + aComponent := projectionTestComponent(t, reflect.TypeOf(fullProjectionOutput{})) + var output []alternateProjectionOutput + + columns, err := view.ProjectionColumnsForOutput(aComponent.View, &output) + + require.NoError(t, err) + require.Equal(t, []string{"campaign_id", "spend"}, columns) +} + +func TestProjectionColumnsForOutput_IgnoresNonSliceOutput(t *testing.T) { + aComponent := projectionTestComponent(t, reflect.TypeOf(fullProjectionOutput{})) + var output struct { + Status string `json:"status"` + } + + columns, err := view.ProjectionColumnsForOutput(aComponent.View, &output) + + require.NoError(t, err) + require.Nil(t, columns) +} + +func TestProjectionColumnsForOutput_FailsForEmptyProjectionDTO(t *testing.T) { + aComponent := projectionTestComponent(t, reflect.TypeOf(fullProjectionOutput{})) + var output []struct{} + + columns, err := view.ProjectionColumnsForOutput(aComponent.View, &output) + + require.Error(t, err) + require.Nil(t, columns) +} + +func TestProjectionColumnsForOutput_UsesSourceAliases(t *testing.T) { + aComponent := projectionTestComponent(t, reflect.TypeOf(fullProjectionOutput{})) + var output []sourceProjectionOutput + + columns, err := view.ProjectionColumnsForOutput(aComponent.View, &output) + + require.NoError(t, err) + require.Equal(t, []string{"alias_value"}, columns) +} + +func TestProjectionColumnsForOutput_IgnoresSkippedAnonymousEmbeds(t *testing.T) { + aComponent := projectionTestComponent(t, reflect.TypeOf(fullProjectionOutput{})) + var output []struct { + ignoredEmbeddedProjection `json:"-"` + Spend int `json:"spend"` + } + + columns, err := view.ProjectionColumnsForOutput(aComponent.View, &output) + + require.NoError(t, err) + require.Equal(t, []string{"spend"}, columns) +} + +func TestProjectionColumnsForOutput_FailsForUnknownField(t *testing.T) { + aComponent := projectionTestComponent(t, reflect.TypeOf(fullProjectionOutput{})) + var output []struct { + Unknown int `json:"unknown"` + } + + columns, err := view.ProjectionColumnsForOutput(aComponent.View, &output) + + require.Error(t, err) + require.Nil(t, columns) +} + +func TestSessionViewProjectionColumns_NarrowsAfterStatePopulation(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + columns := []string{"account_id", "bids"} + aSession := session.New(aComponent.View, session.WithViewProjectionColumns(aComponent.View.Name, columns)) + + err := aSession.SetViewState(context.Background(), aComponent.View) + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.True(t, statelet.Has("account_id")) + require.True(t, statelet.Has("bids")) + require.False(t, statelet.Has("campaign_id")) + + query, err := reader.NewBuilder().CacheSQL(context.Background(), aComponent.View, statelet) + + require.NoError(t, err) + require.Contains(t, query.SQL, "SELECT account_id, bids FROM") + require.Contains(t, query.SQL, "GROUP BY 1") + require.NotContains(t, query.SQL, "campaign_id") +} + +func TestSessionApplyOutputProjectionFromContext_NarrowsChildView(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + var output []struct { + AccountID int `json:"accountId"` + Bids int `json:"bids"` + } + aSession := session.New(aComponent.View) + ctx := ContextWithOutputProjection(context.Background(), &output) + + err := aSession.ApplyOutputProjection(ctx, aComponent.View) + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.Equal(t, []string{"account_id", "bids"}, statelet.Columns) + require.True(t, statelet.Has("account_id")) + require.True(t, statelet.Has("bids")) + require.False(t, statelet.Has("campaign_id")) + + query, err := reader.NewBuilder().Build(context.Background(), reader.WithBuilderView(aComponent.View), reader.WithBuilderStatelet(statelet)) + + require.NoError(t, err) + require.Contains(t, query.SQL, "SELECT account_id, bids FROM") + require.Contains(t, query.SQL, "GROUP BY 1") + require.NotContains(t, query.SQL, "campaign_id") +} + +func TestSessionApplyOutputProjectionFromScopedContext_OnlyAppliesToMatchingView(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + var output []struct { + AccountID int `json:"accountId"` + Bids int `json:"bids"` + } + aSession := session.New(aComponent.View) + ctx := ContextWithViewOutputProjection(context.Background(), "other_view", &output) + + err := aSession.ApplyOutputProjection(ctx, aComponent.View) + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.Empty(t, statelet.Columns) + + ctx = ContextWithViewOutputProjection(context.Background(), aComponent.View.Name, &output) + err = aSession.ApplyOutputProjection(ctx, aComponent.View) + require.NoError(t, err) + require.Equal(t, []string{"account_id", "bids"}, statelet.Columns) +} + +func TestSessionApplyOutputFieldsFromContext_NarrowsChildView(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + aSession := session.New(aComponent.View) + ctx := ContextWithOutputFields(context.Background(), "account_id", "bids") + + err := aSession.ApplyOutputProjection(ctx, aComponent.View) + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.Equal(t, []string{"account_id", "bids"}, statelet.Columns) + require.True(t, statelet.Has("account_id")) + require.True(t, statelet.Has("bids")) + require.False(t, statelet.Has("campaign_id")) +} + +func TestSessionApplyOutputFieldsFromScopedContext_OnlyAppliesToMatchingView(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + aSession := session.New(aComponent.View) + ctx := ContextWithViewOutputFields(context.Background(), "other_view", "account_id", "bids") + + err := aSession.ApplyOutputProjection(ctx, aComponent.View) + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.Empty(t, statelet.Columns) + + ctx = ContextWithViewOutputFields(context.Background(), aComponent.View.Name, "account_id", "bids") + err = aSession.ApplyOutputProjection(ctx, aComponent.View) + require.NoError(t, err) + require.Equal(t, []string{"account_id", "bids"}, statelet.Columns) +} + +func TestSessionApplyOutputProjectionFromOption_NarrowsChildView(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + var output []struct { + AccountID int `json:"accountId"` + Bids int `json:"bids"` + } + aSession := session.New(aComponent.View, session.WithOutputProjection(&output)) + + err := aSession.ApplyOutputProjection(context.Background(), aComponent.View) + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.Equal(t, []string{"account_id", "bids"}, statelet.Columns) +} + +func TestSessionApplyOutputFieldsFromOption_NarrowsChildView(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + aSession := session.New(aComponent.View, session.WithOutputFields("account_id", "bids")) + + err := aSession.ApplyOutputProjection(context.Background(), aComponent.View) + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.Equal(t, []string{"account_id", "bids"}, statelet.Columns) +} + +func TestSessionApplyOutputProjectionFromScopedOption_OnlyAppliesToMatchingView(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + var output []struct { + AccountID int `json:"accountId"` + Bids int `json:"bids"` + } + aSession := session.New(aComponent.View, session.WithViewOutputProjection("other_view", &output)) + + err := aSession.ApplyOutputProjection(context.Background(), aComponent.View) + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.Empty(t, statelet.Columns) + + aSession.Apply(session.WithViewOutputProjection(aComponent.View.Name, &output)) + err = aSession.ApplyOutputProjection(context.Background(), aComponent.View) + require.NoError(t, err) + require.Equal(t, []string{"account_id", "bids"}, statelet.Columns) +} + +func TestSessionApplyOutputFieldsFromScopedOption_OnlyAppliesToMatchingView(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + aSession := session.New(aComponent.View, session.WithViewOutputFields("other_view", "account_id", "bids")) + + err := aSession.ApplyOutputProjection(context.Background(), aComponent.View) + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.Empty(t, statelet.Columns) + + aSession.Apply(session.WithViewOutputFields(aComponent.View.Name, "account_id", "bids")) + err = aSession.ApplyOutputProjection(context.Background(), aComponent.View) + require.NoError(t, err) + require.Equal(t, []string{"account_id", "bids"}, statelet.Columns) +} + +func TestSessionApplyOutputProjectionWithoutHint_LeavesChildViewFullWidth(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + aSession := session.New(aComponent.View) + + err := aSession.ApplyOutputProjection(context.Background(), aComponent.View) + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.Empty(t, statelet.Columns) + + query, err := reader.NewBuilder().Build(context.Background(), reader.WithBuilderView(aComponent.View), reader.WithBuilderStatelet(statelet)) + + require.NoError(t, err) + require.Contains(t, query.SQL, "SELECT t.account_id, t.campaign_id, t.bids FROM") + require.Contains(t, query.SQL, "GROUP BY 1, 2") +} + +func TestSessionApplyOutputProjectionNonSliceHint_DoesNotClearExistingProjection(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + aSession := session.New(aComponent.View) + statelet := aSession.State().Lookup(aComponent.View) + statelet.SetColumns([]string{"account_id", "bids"}) + var output struct { + Status string `json:"status"` + } + ctx := ContextWithOutputProjection(context.Background(), &output) + + err := aSession.ApplyOutputProjection(ctx, aComponent.View) + + require.NoError(t, err) + require.Equal(t, []string{"account_id", "bids"}, statelet.Columns) +} + +func TestSessionApplyOutputProjectionFromContext_FailsForEmptyProjectionDTO(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + var output []struct{} + aSession := session.New(aComponent.View) + ctx := ContextWithOutputProjection(context.Background(), &output) + + err := aSession.ApplyOutputProjection(ctx, aComponent.View) + + require.Error(t, err) + require.Contains(t, err.Error(), "did not map any columns") +} + +func TestSessionApplyOutputProjectionFromContext_FailsForUnknownField(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + var output []struct { + Unknown int `json:"unknown"` + } + aSession := session.New(aComponent.View) + ctx := ContextWithOutputProjection(context.Background(), &output) + + err := aSession.ApplyOutputProjection(ctx, aComponent.View) + + require.Error(t, err) + require.Contains(t, err.Error(), "failed to map output field Unknown") +} + +func TestSessionApplyOutputFieldsFromContext_FailsForUnknownField(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + aSession := session.New(aComponent.View) + ctx := ContextWithOutputFields(context.Background(), "unknown") + + err := aSession.ApplyOutputProjection(ctx, aComponent.View) + + require.Error(t, err) + require.Contains(t, err.Error(), "failed to map output field unknown") +} + +func TestSessionApplyOutputFieldsFromContext_FailsForEmptyFields(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + aSession := session.New(aComponent.View) + ctx := ContextWithOutputFields(context.Background()) + + err := aSession.ApplyOutputProjection(ctx, aComponent.View) + + require.Error(t, err) + require.Contains(t, err.Error(), "did not specify any field names") +} + +func TestSessionApplyOutputFieldsFromOption_FailsForEmptyFields(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + aSession := session.New(aComponent.View, session.WithOutputFields()) + + err := aSession.ApplyOutputProjection(context.Background(), aComponent.View) + + require.Error(t, err) + require.Contains(t, err.Error(), "did not specify any field names") +} + +func TestSessionApplyOutputFieldsFromScopedContext_IgnoresEmptyFieldsForNonMatchingView(t *testing.T) { + aComponent := groupableProjectionTestComponent(t) + aSession := session.New(aComponent.View) + ctx := ContextWithViewOutputFields(context.Background(), "other_view") + + err := aSession.ApplyOutputProjection(ctx, aComponent.View) + + require.NoError(t, err) + statelet := aSession.State().Lookup(aComponent.View) + require.Empty(t, statelet.Columns) +} + +func TestWithOutput_DoesNotEnableProjection(t *testing.T) { + options := newOperateOptions([]OperateOption{WithOutput(&[]alternateProjectionOutput{})}) + + require.NotNil(t, options.output) +} + +func projectionTestComponent(t *testing.T, outputType reflect.Type) *repository.Component { + t.Helper() + aView := view.NewView("projection", "projection", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + &view.Column{Name: "account_id", DataType: "int"}, + &view.Column{Name: "campaign_id", DataType: "int"}, + &view.Column{Name: "impressions", DataType: "int"}, + &view.Column{Name: "spend", DataType: "float"}, + &view.Column{Name: "alias_value", DataType: "int", Tag: `source:"src.alias_value"`}, + }), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + output, err := state.NewType(state.WithSchema(state.NewSchema(outputType))) + require.NoError(t, err) + return &repository.Component{ + View: aView, + Contract: contract.Contract{ + Output: contract.Output{ + Type: *output, + }, + }, + } +} + +func groupableProjectionTestComponent(t *testing.T) *repository.Component { + t.Helper() + aView := view.NewView("groupable_projection", "(SELECT account_id, campaign_id, SUM(bids) AS bids FROM bids GROUP BY 1, 2)", + view.WithConnector(view.NewConnector("test", "sqlite3", ":memory:")), + view.WithColumns(view.Columns{ + &view.Column{Name: "account_id", DataType: "int", Tag: `groupable:"true"`}, + &view.Column{Name: "campaign_id", DataType: "int", Tag: `groupable:"true"`}, + &view.Column{Name: "bids", DataType: "int", Aggregate: true}, + }), + view.WithGroupable(true), + ) + require.NoError(t, aView.Init(context.Background(), view.EmptyResource())) + output, err := state.NewType(state.WithSchema(state.NewSchema(reflect.TypeOf(fullProjectionOutput{})))) + require.NoError(t, err) + return &repository.Component{ + View: aView, + Contract: contract.Contract{ + Output: contract.Output{ + Type: *output, + }, + }, + } +} diff --git a/view/cache.go b/view/cache.go index 3c59bbab..51a4e168 100644 --- a/view/cache.go +++ b/view/cache.go @@ -78,12 +78,13 @@ type ( } CacheInput struct { - Selector *Statelet - Column string - MetaColumn string - IndexMeta bool - Label string - FieldNames []string + Selector *Statelet + Column string + MetaColumn string + IndexMeta bool + Label string + FieldNames []string + StoredFields []ProjectionField } CacheInputFn func() ([]*CacheInput, error) @@ -430,7 +431,10 @@ func (p *ParamValue) clone() *ParamValue { func (c *Cache) GenerateCacheInput(ctx context.Context) ([]*CacheInput, error) { if len(c.Warmup.Cases) == 0 { - input := c.NewInput(NewStatelet()) + input, err := c.newInputWithError(NewStatelet(), nil) + if err != nil { + return nil, err + } if c.maxCasesExceeded(0, 0, input) { if maxCases := c.maxCases(); maxCases > 0 { fmt.Printf("[INFO] cache warmup selector cap view=%s max_cases=%d selected_entries=0 selected_selectors=0\n", c.owner.Name, maxCases) @@ -673,7 +677,10 @@ func (c *Cache) appendSelectors(set *CacheParameters, paramValues [][]interface{ indexes := make([]int, len(paramValues)) generatedEntries := 0 if len(indexes) == 0 { - input := c.newInput(NewStatelet(), set) + input, err := c.newInputWithError(NewStatelet(), set) + if err != nil { + return err + } if c.maxCasesExceeded(selectedEntries, generatedEntries, input) { return nil } @@ -702,7 +709,10 @@ outer: } label := strings.Join(debugParams, ",") - input := c.newInput(selector, set) + input, err := c.newInputWithError(selector, set) + if err != nil { + return err + } input.Label = label if c.maxCasesExceeded(selectedEntries, generatedEntries, input) { return nil @@ -733,18 +743,41 @@ func (c *Cache) NewInput(selector *Statelet) *CacheInput { } func (c *Cache) newInput(selector *Statelet, set *CacheParameters) *CacheInput { + input, err := c.newInputWithError(selector, set) + if err == nil { + return input + } + if c != nil && c.owner != nil { + fmt.Printf("[INFO] cache warmup projection metadata error view=%s field_names=%v error=%v\n", c.owner.Name, c.fieldNamesFor(set), err) + } + return c.newInputWithoutStoredFields(selector, set) +} + +func (c *Cache) newInputWithError(selector *Statelet, set *CacheParameters) (*CacheInput, error) { 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) + storedFields, err := ProjectionFieldsForNames(c.owner, fieldNames) + if err != nil { + return nil, err + } + input := c.newInputWithoutStoredFields(selector, set) + input.StoredFields = append([]ProjectionField(nil), storedFields...) + return input, nil +} + +func (c *Cache) newInputWithoutStoredFields(selector *Statelet, set *CacheParameters) *CacheInput { + fieldNames := c.fieldNamesFor(set) return &CacheInput{ - Selector: selector, - Column: c.Warmup.IndexColumn, - MetaColumn: c.Warmup.IndexColumn, - IndexMeta: (c.Warmup.IndexMeta || c.Warmup.IndexColumn != "") && c.owner.Template.Summary != nil, - FieldNames: append([]string(nil), fieldNames...), + Selector: selector, + Column: c.Warmup.IndexColumn, + MetaColumn: c.Warmup.IndexColumn, + IndexMeta: (c.Warmup.IndexMeta || c.Warmup.IndexColumn != "") && c.owner.Template.Summary != nil, + FieldNames: append([]string(nil), fieldNames...), + StoredFields: nil, } } @@ -758,6 +791,150 @@ func (c *Cache) fieldNamesFor(set *CacheParameters) []string { return c.Warmup.FieldNames } +func (c *Cache) WarmupFieldNamesForSelector(selector *Statelet) ([]string, bool) { + if c == nil || c.Warmup == nil { + return nil, true + } + matchedAny := false + var matchedColumns []string + for _, candidate := range c.Warmup.Cases { + if candidate == nil || len(candidate.FieldNames) == 0 || !c.warmupCaseMatchesSelector(candidate, selector) { + continue + } + columns, ok := c.warmupProjectionColumns(candidate.FieldNames) + if !ok { + return nil, false + } + if !matchedAny { + matchedAny = true + matchedColumns = columns + continue + } + if !stringSlicesEqual(matchedColumns, columns) { + return nil, false + } + } + if matchedAny { + return matchedColumns, true + } + return c.Warmup.FieldNames, true +} + +func (c *Cache) warmupProjectionColumns(fieldNames []string) ([]string, bool) { + if len(fieldNames) == 0 { + return nil, true + } + if c == nil || c.owner == nil { + return fieldNames, true + } + columns, err := ProjectionColumnsForNames(c.owner, fieldNames) + if err == nil { + return columns, true + } + columns = make([]string, 0, len(fieldNames)) + for _, fieldName := range fieldNames { + column, ok := c.owner.ColumnByName(fieldName) + if !ok { + column, ok = c.warmupColumnByNormalizedName(fieldName) + } + if !ok { + return nil, false + } + columns = append(columns, column.Name) + } + return columns, true +} + +func (c *Cache) warmupColumnByNormalizedName(name string) (*Column, bool) { + if c == nil || c.owner == nil { + return nil, false + } + normalized := normalizeProjectionFieldName(name) + for _, column := range c.owner.Columns { + if column == nil { + continue + } + if normalizeProjectionFieldName(column.Name) == normalized || + normalizeProjectionFieldName(column.FieldName()) == normalized || + normalizeProjectionFieldName(column.DatabaseColumn) == normalized { + return column, true + } + } + return nil, false +} + +func stringSlicesEqual(left, right []string) bool { + if len(left) != len(right) { + return false + } + for i := range left { + if normalizeProjectionFieldName(left[i]) != normalizeProjectionFieldName(right[i]) { + return false + } + } + return true +} + +func (c *Cache) warmupCaseMatchesSelector(candidate *CacheParameters, selector *Statelet) bool { + if candidate == nil || selector == nil || selector.Template == nil { + return false + } + for _, paramValue := range candidate.Set { + if paramValue == nil { + return false + } + actual, ok := warmupSelectorValue(selector, paramValue) + if !ok { + return false + } + candidates := paramValue.Values + if paramValue._param != nil { + var err error + candidates, err = c.getParamValues(context.Background(), paramValue) + if err != nil { + return false + } + } + if !warmupValueMatches(actual, candidates) { + return false + } + } + return true +} + +func warmupSelectorValue(selector *Statelet, paramValue *ParamValue) (interface{}, bool) { + if selector == nil || selector.Template == nil || paramValue == nil { + return nil, false + } + if paramValue._param != nil && paramValue._param.Selector() != nil { + stateSelector := paramValue._param.Selector() + if !stateSelector.Has(selector.Template.Pointer()) { + return nil, true + } + return stateSelector.Value(selector.Template.Pointer()), true + } + stateSelector, err := selector.Template.Selector(paramValue.Name) + if err != nil || stateSelector == nil { + return nil, false + } + if !stateSelector.Has(selector.Template.Pointer()) { + return nil, true + } + return stateSelector.Value(selector.Template.Pointer()), true +} + +func warmupValueMatches(actual interface{}, candidates []interface{}) bool { + if len(candidates) == 0 { + return actual == nil || reflect.ValueOf(actual).IsZero() + } + for _, candidate := range candidates { + if reflect.DeepEqual(actual, candidate) || fmt.Sprint(actual) == fmt.Sprint(candidate) { + return true + } + } + return false +} + func (c *Cache) maxCases() int { if c == nil || c.Warmup == nil || c.Warmup.MaxCases == nil || *c.Warmup.MaxCases <= 0 { return 0 diff --git a/view/collector.go b/view/collector.go index b6c5fab5..9e3bc24f 100644 --- a/view/collector.go +++ b/view/collector.go @@ -98,14 +98,26 @@ func normalizeValues(value interface{}) []interface{} { case []string: result := make([]interface{}, 0, len(actual)) for _, item := range actual { + if isBlankStringKey(item) { + continue + } result = append(result, io.NormalizeKey(item)) } return result + case string: + if isBlankStringKey(actual) { + return nil + } + return []interface{}{io.NormalizeKey(value)} default: return []interface{}{io.NormalizeKey(value)} } } +func isBlankStringKey(value string) bool { + return strings.TrimSpace(value) == "" +} + func compositeRows(parts [][]interface{}) [][]interface{} { if len(parts) == 0 { return nil @@ -1143,6 +1155,9 @@ outer: case []string: for j := range actual { + if isBlankStringKey(actual[j]) { + continue + } if _, ok := unique[actual[j]]; ok { continue } @@ -1150,6 +1165,9 @@ outer: result = append(result, actual[j]) } default: + if actual, ok := fieldValue.(string); ok && isBlankStringKey(actual) { + continue + } if count := len(result); count > 0 { if result[count-1] == fieldValue { //value already added continue diff --git a/view/projection.go b/view/projection.go new file mode 100644 index 00000000..eae4b7de --- /dev/null +++ b/view/projection.go @@ -0,0 +1,314 @@ +package view + +import ( + "fmt" + "reflect" + "strings" + + "github.com/viant/sqlx/io/read/cache" +) + +type ProjectionField struct { + Name string `json:",omitempty" yaml:",omitempty"` + FieldName string `json:",omitempty" yaml:",omitempty"` + ColumnName string `json:",omitempty" yaml:",omitempty"` + Source string `json:",omitempty" yaml:",omitempty"` + DimensionKey string `json:",omitempty" yaml:",omitempty"` + MeasureKey string `json:",omitempty" yaml:",omitempty"` + Lookup []string `json:",omitempty" yaml:",omitempty"` +} + +func ProjectionColumnsForOutput(aView *View, output interface{}) ([]string, error) { + if output == nil { + return nil, nil + } + return ProjectionColumnsForType(aView, ProjectionOutputStructType(reflect.TypeOf(output))) +} + +func ProjectionColumnsForNames(aView *View, names []string) ([]string, error) { + if len(names) == 0 { + return nil, nil + } + columns := make([]string, 0, len(names)) + seen := map[string]bool{} + for _, name := range names { + name = strings.TrimSpace(name) + if name == "" { + return nil, fmt.Errorf("output projection for view %s contains empty field name", aView.Name) + } + column, ok := aView.ColumnByName(name) + if !ok { + return nil, fmt.Errorf("failed to map output field %s to view %s column", name, aView.Name) + } + if seen[column.Name] { + continue + } + seen[column.Name] = true + columns = append(columns, column.Name) + } + return columns, nil +} + +func ProjectionFieldsForNames(aView *View, names []string) ([]ProjectionField, error) { + if len(names) == 0 { + return ProjectionFieldsForColumns(aView, aView.Columns), nil + } + fields := make([]ProjectionField, 0, len(names)) + seen := map[string]bool{} + for _, name := range names { + name = strings.TrimSpace(name) + if name == "" { + return nil, fmt.Errorf("output projection for view %s contains empty field name", aView.Name) + } + column, ok := aView.ColumnByName(name) + if !ok { + return nil, fmt.Errorf("failed to map output field %s to view %s column", name, aView.Name) + } + if seen[column.Name] { + continue + } + seen[column.Name] = true + fields = append(fields, ProjectionFieldForViewColumn(aView, column)) + } + return fields, nil +} + +func ProjectionFieldsForColumns(aView *View, columns Columns) []ProjectionField { + if len(columns) == 0 { + return nil + } + fields := make([]ProjectionField, 0, len(columns)) + seen := map[string]bool{} + for _, column := range columns { + if column == nil || column.Name == "" || seen[column.Name] { + continue + } + seen[column.Name] = true + fields = append(fields, ProjectionFieldForViewColumn(aView, column)) + } + return fields +} + +func ProjectionFieldForColumn(column *Column) ProjectionField { + if column == nil { + return ProjectionField{} + } + fieldName := column.FieldName() + if fieldName == "" { + fieldName = column.Name + } + source := projectionFieldSource(column) + return ProjectionField{ + Name: column.Name, + FieldName: fieldName, + ColumnName: column.Name, + Source: source, + Lookup: projectionFieldLookup(column.Name, fieldName, column.DatabaseColumn), + } +} + +func ProjectionFieldForViewColumn(aView *View, column *Column) ProjectionField { + field := ProjectionFieldForColumn(column) + if aView == nil || column == nil || !aView.Groupable { + return field + } + if column.Groupable { + field.MeasureKey = "" + field.DimensionKey = strings.TrimSpace(column.Name) + return field + } + field.DimensionKey = "" + field.MeasureKey = strings.TrimSpace(column.Name) + return field +} + +func SQLXProjectionFields(fields []ProjectionField) []cache.ProjectionField { + if fields == nil { + return nil + } + result := make([]cache.ProjectionField, 0, len(fields)) + for _, field := range fields { + result = append(result, cache.ProjectionField{ + Name: field.Name, + FieldName: field.FieldName, + ColumnName: field.ColumnName, + Source: field.Source, + DimensionKey: field.DimensionKey, + MeasureKey: field.MeasureKey, + Lookup: append([]string(nil), field.Lookup...), + }) + } + return result +} + +func ProjectionColumnsForType(aView *View, rType reflect.Type) ([]string, error) { + if aView == nil || rType == nil { + return nil, nil + } + var result []string + seen := map[string]bool{} + for i := 0; i < rType.NumField(); i++ { + field := rType.Field(i) + if field.PkgPath != "" { + continue + } + if skipProjectionField(field) { + continue + } + if field.Anonymous { + if nested := ProjectionStructType(field.Type); nested != nil { + columns, err := ProjectionColumnsForType(aView, nested) + if err != nil { + return nil, err + } + for _, column := range columns { + if !seen[column] { + result = append(result, column) + seen[column] = true + } + } + continue + } + } + column, err := projectionColumnForField(aView, field) + if err != nil { + return nil, err + } + if !seen[column.Name] { + result = append(result, column.Name) + seen[column.Name] = true + } + } + if len(result) == 0 { + return nil, fmt.Errorf("output projection for view %s did not map any columns", aView.Name) + } + return result, nil +} + +func ProjectionStructType(rType reflect.Type) reflect.Type { + if rType == nil { + return nil + } + for rType.Kind() == reflect.Ptr { + rType = rType.Elem() + } + if rType.Kind() == reflect.Slice || rType.Kind() == reflect.Array { + rType = rType.Elem() + for rType.Kind() == reflect.Ptr { + rType = rType.Elem() + } + } + if rType.Kind() != reflect.Struct { + return nil + } + return rType +} + +func ProjectionOutputStructType(rType reflect.Type) reflect.Type { + if rType == nil { + return nil + } + for rType.Kind() == reflect.Ptr { + rType = rType.Elem() + } + if rType.Kind() != reflect.Slice && rType.Kind() != reflect.Array { + return nil + } + return ProjectionStructType(rType) +} + +func projectionColumnForField(aView *View, field reflect.StructField) (*Column, error) { + for _, candidate := range projectionFieldCandidates(field) { + if column, ok := aView.ColumnByName(candidate); ok { + return column, nil + } + } + return nil, fmt.Errorf("failed to map output field %s to a column in view %s", field.Name, aView.Name) +} + +func projectionFieldCandidates(field reflect.StructField) []string { + var result []string + add := func(value string) { + value = strings.TrimSpace(value) + if value == "" || value == "-" { + return + } + for _, item := range strings.Split(value, "|") { + item = strings.TrimSpace(item) + if item == "" || item == "-" { + continue + } + result = append(result, item) + } + } + add(tagName(field.Tag.Get("sqlx"))) + add(tagName(field.Tag.Get("source"))) + add(tagName(field.Tag.Get("json"))) + add(field.Name) + return result +} + +func skipProjectionField(field reflect.StructField) bool { + return tagName(field.Tag.Get("json")) == "-" || tagName(field.Tag.Get("sqlx")) == "-" +} + +func tagName(tag string) string { + if index := strings.Index(tag, ","); index != -1 { + tag = tag[:index] + } + return strings.TrimSpace(tag) +} + +func projectionFieldSource(column *Column) string { + if column == nil || column.Tag == "" { + return "" + } + return reflect.StructTag(column.Tag).Get("source") +} + +func projectionFieldLookup(values ...string) []string { + seen := map[string]bool{} + var result []string + for _, value := range values { + for _, candidate := range projectionFieldLookupCandidates(value) { + if seen[candidate] { + continue + } + seen[candidate] = true + result = append(result, candidate) + } + } + return result +} + +func projectionFieldLookupCandidates(value string) []string { + value = strings.TrimSpace(value) + if value == "" { + return nil + } + values := []string{value} + if index := strings.LastIndex(value, "."); index != -1 && index+1 < len(value) { + values = append(values, value[index+1:]) + } + result := make([]string, 0, len(values)*3) + for _, candidate := range values { + result = append(result, candidate) + normalized := normalizeProjectionFieldName(candidate) + if normalized != "" && normalized != candidate { + result = append(result, normalized) + } + lower := strings.ToLower(candidate) + if lower != candidate && lower != normalized { + result = append(result, lower) + } + } + return result +} + +func normalizeProjectionFieldName(value string) string { + value = strings.TrimSpace(strings.ToLower(value)) + value = strings.ReplaceAll(value, "_", "") + value = strings.ReplaceAll(value, "-", "") + value = strings.ReplaceAll(value, ".", "") + return value +} diff --git a/view/projection_test.go b/view/projection_test.go new file mode 100644 index 00000000..9216041c --- /dev/null +++ b/view/projection_test.go @@ -0,0 +1,78 @@ +package view + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestProjectionFieldForColumn_DoesNotSetGroupedSemanticKeys(t *testing.T) { + dimension := ProjectionFieldForColumn(&Column{Name: "audience_id", Groupable: true}) + + assert.Empty(t, dimension.DimensionKey) + assert.Empty(t, dimension.MeasureKey) + + measure := ProjectionFieldForColumn(&Column{Name: "spend", Aggregate: true}) + + assert.Empty(t, measure.DimensionKey) + assert.Empty(t, measure.MeasureKey) +} + +func TestProjectionFieldForColumn_KeepsSourceOutOfLookup(t *testing.T) { + field := ProjectionFieldForColumn(&Column{ + Name: "campaign_id", + DatabaseColumn: "CAMPAIGN_ID", + Tag: `source:"ID"`, + }) + + assert.Equal(t, "ID", field.Source) + assert.NotContains(t, field.Lookup, "ID") + assert.NotContains(t, field.Lookup, "id") + assert.Contains(t, field.Lookup, "campaign_id") + assert.Contains(t, field.Lookup, "campaignid") +} + +func TestProjectionFieldsForNames_NonGroupedViewDoesNotSetGroupedSemanticKeys(t *testing.T) { + aView := NewView("events", "events", + WithConnector(NewConnector("test", "sqlite3", ":memory:")), + WithColumns(Columns{ + {Name: "order_id", DataType: "int", Groupable: true}, + {Name: "spend", DataType: "float", Aggregate: true}, + }), + ) + require.NoError(t, aView.Init(context.Background(), EmptyResource())) + fields, err := ProjectionFieldsForNames(aView, []string{"order_id", "spend"}) + require.NoError(t, err) + require.Len(t, fields, 2) + + assert.Empty(t, fields[0].DimensionKey) + assert.Empty(t, fields[0].MeasureKey) + assert.Empty(t, fields[1].DimensionKey) + assert.Empty(t, fields[1].MeasureKey) +} + +func TestProjectionFieldsForNames_GroupedViewTreatsNonGroupableColumnsAsMeasures(t *testing.T) { + aView := NewView("linePeriodSummary", "line_period_summary", + WithConnector(NewConnector("test", "sqlite3", ":memory:")), + WithGroupable(true), + WithColumns(Columns{ + {Name: "audience_id", DataType: "int", Groupable: true}, + {Name: "SPEND", DataType: "float"}, + {Name: "PERIOD_ECPM", DataType: "float"}, + }), + ) + require.NoError(t, aView.Init(context.Background(), EmptyResource())) + fields, err := ProjectionFieldsForNames(aView, []string{"audience_id", "spend", "period_ecpm"}) + require.NoError(t, err) + require.Len(t, fields, 3) + + assert.Equal(t, "audience_id", fields[0].DimensionKey) + + assert.Empty(t, fields[1].DimensionKey) + assert.Equal(t, "SPEND", fields[1].MeasureKey) + + assert.Empty(t, fields[2].DimensionKey) + assert.Equal(t, "PERIOD_ECPM", fields[2].MeasureKey) +} diff --git a/view/state.go b/view/state.go index 48761aa7..7aaee716 100644 --- a/view/state.go +++ b/view/state.go @@ -73,6 +73,11 @@ func (s *Statelet) Add(fieldName string, isHolder bool) { } } +func (s *Statelet) SetColumns(columns []string) { + s.Columns = append([]string(nil), columns...) + s._columnNames = Names(s.Columns).Index() +} + // AppendFilters safely appends filters to the selector's Filters to avoid data races. func (s *Statelet) AppendFilters(filters predicate.Filters) { if len(filters) == 0 { diff --git a/warmup/cache.go b/warmup/cache.go index a3a3ec38..46f39d8f 100644 --- a/warmup/cache.go +++ b/warmup/cache.go @@ -171,6 +171,7 @@ func (c *matchersCollector) createIndexWarmupEntry(ctx context.Context, aView *v } return } + build.StoredFields = view.SQLXProjectionFields(cacheInput.StoredFields) aChan <- func() (*warmupEntry, error) { return &warmupEntry{ diff --git a/warmup/cache_test.go b/warmup/cache_test.go index 9b1f7c79..b7334521 100644 --- a/warmup/cache_test.go +++ b/warmup/cache_test.go @@ -244,6 +244,7 @@ Connectors: Views: - Name: events + Groupable: true Connector: Ref: db Table: events @@ -283,6 +284,10 @@ Views: require.NoError(t, err) require.NotEmpty(t, fieldInput) assert.Equal(t, []string{"Quantity"}, fieldInput[0].FieldNames) + require.Len(t, fieldInput[0].StoredFields, 1) + assert.Equal(t, "quantity", fieldInput[0].StoredFields[0].Name) + assert.Equal(t, "quantity", fieldInput[0].StoredFields[0].FieldName) + assert.Contains(t, fieldInput[0].StoredFields[0].Lookup, "quantity") fieldQuery, err := builder.CacheSQL(context.Background(), aView, fieldInput[0].Selector) require.NoError(t, err) @@ -294,6 +299,321 @@ Views: assert.Contains(t, fieldQuery.SQL, "quantity") } +func TestGenerateCacheInput_StoresDefaultProjectionFieldMetadata(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 + Groupable: true + Connector: + Ref: db + Table: events + Columns: + - Name: event_type_id + DataType: int + Tag: 'source:"e.event_type_id"' + Groupable: true + - Name: quantity + DataType: int + Aggregate: true + Cache: + Ref: aerospike + Warmup: + IndexColumn: event_type_id + Selector: + Constraints: + Projection: true + 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.NotEmpty(t, input) + require.Len(t, input[0].StoredFields, 2) + + assert.Equal(t, "event_type_id", input[0].StoredFields[0].Name) + assert.Equal(t, "event_type_id", input[0].StoredFields[0].DimensionKey) + assert.Empty(t, input[0].StoredFields[0].MeasureKey) + assert.NotContains(t, input[0].StoredFields[0].Lookup, "e.event_type_id") + assert.Contains(t, input[0].StoredFields[0].Lookup, "event_type_id") + + assert.Equal(t, "quantity", input[0].StoredFields[1].Name) + assert.Empty(t, input[0].StoredFields[1].DimensionKey) + assert.Equal(t, "quantity", input[0].StoredFields[1].MeasureKey) +} + +func TestGenerateCacheInput_ReturnsStoredFieldMetadataError(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 + Groupable: true + Connector: + Ref: db + Table: events + Columns: + - Name: event_type_id + DataType: int + - Name: quantity + DataType: int + Cache: + Ref: aerospike + Warmup: + IndexColumn: event_type_id + Selector: + Constraints: + Projection: true + 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] + + aView.Cache.Warmup.FieldNames = []string{"missing"} + _, err = aView.Cache.GenerateCacheInput(context.Background()) + + require.Error(t, err) + assert.Contains(t, err.Error(), "failed to map output field missing") +} + +func TestCacheNewInput_ReturnsInputWhenStoredFieldMetadataFails(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 + Groupable: true + Connector: + Ref: db + Table: events + Columns: + - Name: event_type_id + DataType: int + Cache: + Ref: aerospike + Warmup: + IndexColumn: event_type_id + Selector: + Constraints: + Projection: true + 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] + aView.Cache.Warmup.FieldNames = []string{"missing"} + + input := aView.Cache.NewInput(view.NewStatelet()) + + require.NotNil(t, input) + assert.Equal(t, []string{"missing"}, input.FieldNames) + assert.Empty(t, input.StoredFields) +} + +func TestSQLXProjectionFieldsCopiesStoredFieldMetadata(t *testing.T) { + actual := view.SQLXProjectionFields([]view.ProjectionField{ + { + Name: "order_id", + FieldName: "OrderId", + ColumnName: "order_id", + Source: "o.order_id", + DimensionKey: "order_id", + Lookup: []string{"order_id", "OrderId", "orderid"}, + }, + { + Name: "bids", + FieldName: "Bids", + MeasureKey: "bids", + Lookup: []string{"bids", "Bids"}, + }, + }) + + require.Len(t, actual, 2) + assert.Equal(t, "order_id", actual[0].Name) + assert.Equal(t, "OrderId", actual[0].FieldName) + assert.Equal(t, "order_id", actual[0].ColumnName) + assert.Equal(t, "o.order_id", actual[0].Source) + assert.Equal(t, "order_id", actual[0].DimensionKey) + assert.Empty(t, actual[0].MeasureKey) + assert.Equal(t, []string{"order_id", "OrderId", "orderid"}, actual[0].Lookup) + + assert.Equal(t, "bids", actual[1].Name) + assert.Empty(t, actual[1].DimensionKey) + assert.Equal(t, "bids", actual[1].MeasureKey) +} + +func TestCreateIndexWarmupEntrySetsMatcherStoredFields(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 + Groupable: true + Connector: + Ref: db + Table: events + Columns: + - Name: event_type_id + DataType: int + Groupable: true + - Name: quantity + DataType: int + Aggregate: true + Cache: + Ref: aerospike + Warmup: + IndexColumn: event_type_id + FieldNames: + - quantity + Selector: + Constraints: + Projection: true + 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] + inputs, err := aView.Cache.GenerateCacheInput(context.Background()) + require.NoError(t, err) + require.NotEmpty(t, inputs) + + collector := make(chan warmupEntryFn, 1) + (&matchersCollector{builder: reader.NewBuilder(), view: aView}).createIndexWarmupEntry(context.Background(), aView, collector, inputs[0]) + entry, err := (<-collector)() + + require.NoError(t, err) + require.NotNil(t, entry.matcher) + require.Len(t, entry.matcher.StoredFields, 1) + assert.Equal(t, "quantity", entry.matcher.StoredFields[0].Name) + assert.Equal(t, "quantity", entry.matcher.StoredFields[0].MeasureKey) +} + +func TestCreateMetaWarmupEntryDoesNotSetDataStoredFields(t *testing.T) { + dbPath := path.Join(t.TempDir(), "events.db") + db, err := sql.Open("sqlite3", dbPath) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, db.Close()) + }) + _, err = db.Exec(`CREATE TABLE EVENTS (event_type_id INTEGER, quantity INTEGER)`) + require.NoError(t, err) + + 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: "`+dbPath+`" + +Views: + - Name: events + Connector: + Ref: db + Table: events + Columns: + - Name: event_type_id + DataType: int + Groupable: true + - Name: quantity + DataType: int + Aggregate: true + Cache: + Ref: aerospike + Warmup: + IndexColumn: event_type_id + FieldNames: + - quantity + Selector: + Constraints: + Projection: true + Template: + Summary: + Name: EventsMeta + Source: 'SELECT COUNT(*) AS TOTAL_RECORDS, event_type_id FROM ($View.Expand($criteria)) GROUP BY event_type_id' + 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] + inputs, err := aView.Cache.GenerateCacheInput(context.Background()) + require.NoError(t, err) + require.NotEmpty(t, inputs) + require.NotEmpty(t, inputs[0].StoredFields) + + collector := make(chan warmupEntryFn, 1) + (&matchersCollector{builder: reader.NewBuilder(), view: aView}).createMetaWarmupEntry(context.Background(), aView, collector, inputs[0]) + entry, err := (<-collector)() + + require.NoError(t, err) + require.NotNil(t, entry.matcher) + assert.Empty(t, entry.matcher.StoredFields) +} + func TestGenerateCacheInput_AppliesWarmupLimitOverride(t *testing.T) { resourcePath := path.Join(t.TempDir(), "resource.yaml") require.NoError(t, os.WriteFile(resourcePath, []byte(`