diff --git a/compose/service/record.go b/compose/service/record.go index d7e7508fd..fc263560a 100644 --- a/compose/service/record.go +++ b/compose/service/record.go @@ -304,65 +304,7 @@ func (svc record) Report(ctx context.Context, namespaceID, moduleID uint64, metr return RecordErrNotAllowedToSearch() } - // Map dimension to the aggregate group - // @note we only ever used a single dimension so this is ok - dim := []dal.AttributeMapping{ - dal.SimpleAttr{ - Ident: "dimension_0", - Expr: dimensions, - }, - } - - // Map metrics to the aggregate attrs - // - count is always present - mms := []dal.AttributeMapping{ - dal.SimpleAttr{ - Ident: "count", - Expr: "count(ID)", - Props: dal.MapProperties{ - Type: dal.TypeNumber{}, - }, - }, - } - - // - other requested metrices - if len(metrics) > 0 { - pts := strings.Split(metrics, " AS ") - expr := strings.TrimSpace(pts[0]) - ident := expr - if len(pts) > 1 { - ident = strings.TrimSpace(pts[1]) - } - - mms = append(mms, dal.SimpleAttr{ - Ident: ident, - Expr: expr, - Props: dal.MapProperties{ - Type: dal.TypeNumber{}, - }, - }) - } - - // Build the pipeline - pp := dal.Pipeline{ - &dal.Datasource{ - Ident: "ds", - Filter: filter.Generic(filter.WithExpression(f)), - ModelRef: dal.ModelRef{ - ConnectionID: m.Config.DAL.ConnectionID, - ResourceID: m.ID, - ResourceType: types.ModuleResourceType, - }, - }, - &dal.Aggregate{ - Ident: "agg", - RelSource: "ds", - Group: dim, - OutAttributes: mms, - }, - } - - err = pp.LinkSteps() + pp, err := recordReportToDalPipeline(m, metrics, dimensions, f) if err != nil { return err } @@ -1753,6 +1695,68 @@ func loadRecord(ctx context.Context, s store.Storer, namespaceID, moduleID, reco return } +func recordReportToDalPipeline(m *types.Module, metrics, dimensions, f string) (pp dal.Pipeline, err error) { + // Map dimension to the aggregate group + // @note we only ever used a single dimension so this is ok + dim := []dal.AttributeMapping{ + dal.SimpleAttr{ + Ident: "dimension_0", + Expr: dimensions, + }, + } + + // Map metrics to the aggregate attrs + // - count is always present + mms := []dal.AttributeMapping{ + dal.SimpleAttr{ + Ident: "count", + Expr: "count(ID)", + Props: dal.MapProperties{ + Type: dal.TypeNumber{}, + }, + }, + } + + // - other requested metrices + if len(metrics) > 0 { + pts := strings.Split(metrics, " AS ") + expr := strings.TrimSpace(pts[0]) + ident := expr + if len(pts) > 1 { + ident = strings.TrimSpace(pts[1]) + } + + mms = append(mms, dal.SimpleAttr{ + Ident: ident, + Expr: expr, + Props: dal.MapProperties{ + Type: dal.TypeNumber{}, + }, + }) + } + + // Build the pipeline + pp = dal.Pipeline{ + &dal.Datasource{ + Ident: "ds", + Filter: filter.Generic(filter.WithExpression(f)), + ModelRef: dal.ModelRef{ + ConnectionID: m.Config.DAL.ConnectionID, + ResourceID: m.ID, + ResourceType: types.ModuleResourceType, + }, + }, + &dal.Aggregate{ + Ident: "agg", + RelSource: "ds", + Group: dim, + OutAttributes: mms, + }, + } + + return pp, pp.LinkSteps() +} + func (ei ErrorIndex) Add(err string) { if _, has := ei[err]; has { ei[err]++ diff --git a/compose/service/record_test.go b/compose/service/record_test.go index b96bf4674..48a44e0e6 100644 --- a/compose/service/record_test.go +++ b/compose/service/record_test.go @@ -1014,3 +1014,45 @@ func TestSetRecordOwner(t *testing.T) { req.Equal(upd.OwnedBy, invoker.ID) }) } + +func TestRecordReportToDalPipeline(t *testing.T) { + mod := &types.Module{ + ID: 10, + NamespaceID: 42, + } + + t.Run("no additional metrics", func(t *testing.T) { + pp, err := recordReportToDalPipeline(mod, "", "created_at", "") + require.NoError(t, err) + + require.Len(t, pp, 2) + + agg := pp[1].(*dal.Aggregate) + require.Len(t, agg.OutAttributes, 1) + require.Equal(t, "count", agg.OutAttributes[0].Identifier()) + }) + + t.Run("additional metrics with alias", func(t *testing.T) { + pp, err := recordReportToDalPipeline(mod, "MAX(numbers) AS something", "created_at", "") + require.NoError(t, err) + + require.Len(t, pp, 2) + + agg := pp[1].(*dal.Aggregate) + require.Len(t, agg.OutAttributes, 2) + require.Equal(t, "something", agg.OutAttributes[1].Identifier()) + require.Equal(t, "MAX(numbers)", agg.OutAttributes[1].Expression()) + }) + + t.Run("additional metrics without alias", func(t *testing.T) { + pp, err := recordReportToDalPipeline(mod, "MAX(numbers)", "created_at", "") + require.NoError(t, err) + + require.Len(t, pp, 2) + + agg := pp[1].(*dal.Aggregate) + require.Len(t, agg.OutAttributes, 2) + require.Equal(t, "MAX(numbers)", agg.OutAttributes[1].Identifier()) + require.Equal(t, "MAX(numbers)", agg.OutAttributes[1].Expression()) + }) +} diff --git a/system/rest/report.go b/system/rest/report.go index d4485c232..f639b74f1 100644 --- a/system/rest/report.go +++ b/system/rest/report.go @@ -28,7 +28,7 @@ type ( Delete(ctx context.Context, ID uint64) (err error) Undelete(ctx context.Context, ID uint64) (err error) Run(ctx context.Context, ID uint64, dd reporting.FrameDefinitionSet) (rr []*reporting.Frame, err error) - Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out reporting.[]FrameDescription, err error) + Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out []reporting.FrameDescription, err error) } reportAccessController interface { diff --git a/system/service/report.go b/system/service/report.go index 7f0805419..23d046783 100644 --- a/system/service/report.go +++ b/system/service/report.go @@ -293,8 +293,8 @@ func (svc *report) Undelete(ctx context.Context, ID uint64) (err error) { } // @todo actionlog? -func (svc *report) Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out reporting.[]FrameDescription, err error) { - out = make(reporting.[]FrameDescription, 0, len(sources)*2) +func (svc *report) Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out []reporting.FrameDescription, err error) { + out = make([]reporting.FrameDescription, 0, len(sources)*2) err = func() (err error) { if !svc.ac.CanCreateReport(ctx) {