Port the old system reporter to the DAL pipeline

compose/service/record.go

@@ -4,12 +4,13 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/cortezaproject/corteza-server/pkg/revisions"
 	"regexp"
 	"sort"
 	"strconv"
 	"time"
 
+	"github.com/cortezaproject/corteza-server/pkg/revisions"
+
 	"github.com/cortezaproject/corteza-server/pkg/dal"
 	"github.com/cortezaproject/corteza-server/pkg/locale"
 
@@ -23,7 +24,6 @@ import (
 	"github.com/cortezaproject/corteza-server/pkg/envoy/resource"
 	"github.com/cortezaproject/corteza-server/pkg/errors"
 	"github.com/cortezaproject/corteza-server/pkg/eventbus"
-	"github.com/cortezaproject/corteza-server/pkg/report"
 	"github.com/cortezaproject/corteza-server/store"
 )
 
@@ -112,8 +112,6 @@ type (
 		RecordExport(context.Context, types.RecordFilter) error
 		RecordImport(context.Context, error) error
 
-		Datasource(context.Context, *report.LoadStepDefinition) (report.Datasource, error)
-
 		Create(ctx context.Context, record *types.Record) (*types.Record, error)
 		Update(ctx context.Context, record *types.Record) (*types.Record, error)
 		Bulk(ctx context.Context, oo ...*types.RecordBulkOperation) (types.RecordSet, error)

compose/service/record_datasource.go

@@ -1,146 +0,0 @@
-package service
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/cortezaproject/corteza-server/compose/types"
-	"github.com/cortezaproject/corteza-server/pkg/errors"
-	"github.com/cortezaproject/corteza-server/pkg/report"
-	"github.com/cortezaproject/corteza-server/store"
-	"github.com/spf13/cast"
-)
-
-func (svc record) Datasource(ctx context.Context, ld *report.LoadStepDefinition) (report.Datasource, error) {
-	var (
-		moduleID    uint64
-		namespaceID uint64
-		err         error
-
-		def = ld.Definition
-	)
-
-	if mr, ok := def["namespaceID"]; ok {
-		namespaceID, err = cast.ToUint64E(mr)
-		if err != nil {
-			return nil, err
-		}
-	} else if mr, ok = def["namespace"]; ok {
-		// slug; fetch from store
-		ns, err := store.LookupComposeNamespaceBySlug(ctx, svc.store, mr.(string))
-		if errors.IsNotFound(err) {
-			err = NamespaceErrNotFound()
-		}
-		if err != nil {
-			return nil, err
-		}
-		namespaceID = ns.ID
-	} else {
-		return nil, fmt.Errorf("compose namespace not defined")
-	}
-
-	// check if namespace exists
-	ns, err := store.LookupComposeNamespaceByID(ctx, svc.store, namespaceID)
-	if err != nil {
-		return nil, err
-	}
-	if ns.DeletedAt != nil {
-		return nil, NamespaceErrNotFound()
-	}
-
-	if mr, ok := def["moduleID"]; ok {
-		moduleID, err = cast.ToUint64E(mr)
-		if err != nil {
-			return nil, err
-		}
-	} else if mr, ok = def["module"]; ok {
-		// handle; fetch from store
-		mod, err := store.LookupComposeModuleByNamespaceIDHandle(ctx, svc.store, namespaceID, mr.(string))
-		if errors.IsNotFound(err) {
-			err = ModuleErrNotFound()
-		}
-		if err != nil {
-			return nil, err
-		}
-		moduleID = mod.ID
-	}
-
-	// Find mod
-	mod, err := store.LookupComposeModuleByID(ctx, svc.store, moduleID)
-	if err != nil {
-		return nil, err
-	}
-	mod.Fields, _, err = store.SearchComposeModuleFields(ctx, svc.store, types.ModuleFieldFilter{
-		ModuleID: []uint64{mod.ID},
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	if len(ld.Columns) == 0 {
-		cols := make(report.FrameColumnSet, 0, len(mod.Fields)+8)
-
-		var c *report.FrameColumn
-		c = report.MakeColumnOfKind("Record")
-		c.System = true
-		c.Name = "id"
-		c.Label = "Record ID"
-		c.Primary = true
-		c.Unique = true
-		cols = append(cols, c)
-
-		for _, f := range mod.Fields {
-			k := f.Kind
-			c = report.MakeColumnOfKind(k)
-			c.Name = f.Name
-			c.Label = f.Label
-			c.Options = f.Options
-			if c.Label == "" {
-				c.Label = c.Name
-			}
-			cols = append(cols, c)
-		}
-
-		// Sys fields
-		c = report.MakeColumnOfKind("DateTime")
-		c.System = true
-		c.Name = "createdAt"
-		c.Label = "Created at"
-		cols = append(cols, c)
-
-		c = report.MakeColumnOfKind("User")
-		c.System = true
-		c.Name = "createdBy"
-		c.Label = "Created by"
-		cols = append(cols, c)
-
-		c = report.MakeColumnOfKind("DateTime")
-		c.System = true
-		c.Name = "updatedAt"
-		c.Label = "Updated at"
-		cols = append(cols, c)
-
-		c = report.MakeColumnOfKind("User")
-		c.System = true
-		c.Name = "updatedBy"
-		c.Label = "Updated by"
-		cols = append(cols, c)
-
-		c = report.MakeColumnOfKind("DateTime")
-		c.System = true
-		c.Name = "deletedAt"
-		c.Label = "Deleted at"
-		cols = append(cols, c)
-
-		c = report.MakeColumnOfKind("User")
-		c.System = true
-		c.Name = "deletedBy"
-		c.Label = "Deleted by"
-		cols = append(cols, c)
-
-		ld.Columns = cols
-	}
-
-	return nil, fmt.Errorf("@todo pending migration to DAL")
-	// return store.ComposeRecordDatasource(ctx, svc.store, mod, ld)
-}

pkg/dal/service.go

@@ -445,23 +445,23 @@ func (svc *service) datasource(ctx context.Context, mf ModelRef, f filter.Filter
 	return iter, model, err
 }
 
-// RunPipeline returns an iterator based on the provided Pipeline
+// Run returns an iterator based on the provided Pipeline
 // @todo consider moving the Search method to utilize this also
-func (svc *service) RunPipeline(ctx context.Context, pp Pipeline) (iter Iterator, err error) {
-	return svc.runPipeline(ctx, pp.root(), false)
+func (svc *service) Run(ctx context.Context, pp Pipeline) (iter Iterator, err error) {
+	return svc.run(ctx, pp.root(), false)
 }
 
-// DryrunPipeline processes the Pipeline but doesn't return the iterator
+// Dryrun processes the Pipeline but doesn't return the iterator
 // @todo consider reworking this; I'm not the biggest fan of this one
 //
 // The method is primarily used by system reports to obtain some metadata
-func (svc *service) DryrunPipeline(ctx context.Context, pp Pipeline) (err error) {
-	_, err = svc.runPipeline(ctx, pp.root(), true)
+func (svc *service) Dryrun(ctx context.Context, pp Pipeline) (err error) {
+	_, err = svc.run(ctx, pp.root(), true)
 	return
 }
 
-// runPipeline is the recursive counterpart to RunPipeline
-func (svc *service) runPipeline(ctx context.Context, s PipelineStep, dry bool) (it Iterator, err error) {
+// run is the recursive counterpart to run
+func (svc *service) run(ctx context.Context, s PipelineStep, dry bool) (it Iterator, err error) {
 	switch s := s.(type) {
 	case *Datasource:
 		err = s.init(ctx, svc.datasource)
@@ -474,7 +474,7 @@ func (svc *service) runPipeline(ctx context.Context, s PipelineStep, dry bool) (
 		return s.exec(ctx)
 
 	case *Aggregate:
-		it, err = svc.runPipeline(ctx, s.rel, dry)
+		it, err = svc.run(ctx, s.rel, dry)
 		if err != nil {
 			return
 		}
@@ -492,11 +492,11 @@ func (svc *service) runPipeline(ctx context.Context, s PipelineStep, dry bool) (
 	case *Join:
 		var left Iterator
 		var right Iterator
-		left, err = svc.runPipeline(ctx, s.relLeft, dry)
+		left, err = svc.run(ctx, s.relLeft, dry)
 		if err != nil {
 			return
 		}
-		right, err = svc.runPipeline(ctx, s.relRight, dry)
+		right, err = svc.run(ctx, s.relRight, dry)
 		if err != nil {
 			return
 		}
@@ -513,11 +513,11 @@ func (svc *service) runPipeline(ctx context.Context, s PipelineStep, dry bool) (
 	case *Link:
 		var left Iterator
 		var right Iterator
-		left, err = svc.runPipeline(ctx, s.relLeft, dry)
+		left, err = svc.run(ctx, s.relLeft, dry)
 		if err != nil {
 			return
 		}
-		right, err = svc.runPipeline(ctx, s.relRight, dry)
+		right, err = svc.run(ctx, s.relRight, dry)
 		if err != nil {
 			return
 		}

pkg/reportutils/frame_builder.go

@@ -0,0 +1,113 @@
+package reportutils
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/cortezaproject/corteza-server/pkg/dal"
+	"github.com/cortezaproject/corteza-server/pkg/filter"
+	"github.com/cortezaproject/corteza-server/system/types"
+)
+
+type (
+	// reportFrameBuilder is a helper struct for building report frames from
+	// dal iterators
+	reportFrameBuilder struct {
+		def         *types.ReportFrameDefinition
+		frame       *types.ReportFrame
+		attrMapping map[string]int
+	}
+)
+
+func newReportFrameBuilder(def *types.ReportFrameDefinition) *reportFrameBuilder {
+	// Index requested columns for easier lookup
+	attrMap := make(map[string]int)
+	for i, c := range def.Columns {
+		attrMap[c.Name] = i
+	}
+
+	out := &reportFrameBuilder{
+		def:         def,
+		attrMapping: attrMap,
+	}
+
+	out.freshFrame()
+
+	return out
+}
+
+// withRefs includes additional metadata required by the link step
+func (b *reportFrameBuilder) withRefs(col string) {
+	b.frame.RelColumn = col
+}
+
+// addRow adds a new dal.Row to the frame
+func (b *reportFrameBuilder) addRow(r *dal.Row) {
+	rrow := make(types.ReportFrameRow, len(b.def.Columns))
+
+	for k, cc := range r.CountValues() {
+		ix, ok := b.attrMapping[k]
+		if !ok {
+			continue
+		}
+
+		auxRow := make(types.ReportFrameRow, cc)
+		for i := uint(0); i < cc; i++ {
+			val, _ := r.GetValue(k, i)
+			auxRow[i] = b.stringifyVal(k, val)
+		}
+		rrow[ix] = b.joinMultiVal(k, auxRow)
+	}
+
+	b.frame.Rows = append(b.frame.Rows, rrow)
+
+	if b.frame.RelColumn != "" {
+		v, _ := r.GetValue(b.frame.RelColumn, 0)
+		b.frame.RefValue = b.stringifyVal(b.frame.RelColumn, v)
+	}
+}
+
+func (b *reportFrameBuilder) stringifyVal(col string, val any) string {
+	// @todo nicer formatting and such? V1 didn't do much different
+	return fmt.Sprintf("%v", val)
+}
+
+func (b *reportFrameBuilder) joinMultiVal(col string, vals []string) string {
+	// @todo add delimiter (extend attrs)
+	return strings.Join(vals, "\n")
+}
+
+// done returns the constructed frame and prepares a new frame with the same
+// metadata as the original one
+func (b *reportFrameBuilder) done() *types.ReportFrame {
+	out := b.frame
+	b.freshFrame()
+
+	return out
+}
+
+func (b *reportFrameBuilder) freshFrame() {
+	if b.frame != nil {
+		aux := *b.frame
+		b.frame = &aux
+		b.frame.Rows = nil
+		return
+	}
+
+	b.frame = &types.ReportFrame{
+		Name:   b.def.Name,
+		Source: b.def.Source,
+		Ref:    b.def.Ref,
+
+		Columns: b.def.Columns,
+		Sort:    b.def.Sort,
+		Filter:  b.def.Filter,
+	}
+
+	if b.def.Paging != nil {
+		b.frame.Paging = &filter.Paging{
+			Limit:      b.def.Paging.Limit,
+			PageCursor: b.def.Paging.PageCursor,
+		}
+	}
+}

pkg/reportutils/reportutils.go

@@ -0,0 +1,529 @@
+package reportutils
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cortezaproject/corteza-server/pkg/dal"
+	"github.com/cortezaproject/corteza-server/pkg/filter"
+	"github.com/cortezaproject/corteza-server/system/types"
+	"github.com/spf13/cast"
+)
+
+type (
+	ReportWorkload struct {
+		Pipeline  dal.Pipeline
+		FrameDefs types.ReportFrameDefinitionSet
+	}
+
+	PipelineRunner interface {
+		FindModel(dal.ModelRef) *dal.Model
+		Run(context.Context, dal.Pipeline) (dal.Iterator, error)
+		Dryrun(context.Context, dal.Pipeline) error
+	}
+)
+
+// Workloads creates a set of workloads for the given pipeline and frame definitions
+func Workloads(pr PipelineRunner, steps types.ReportStepSet, defs types.ReportFrameDefinitionSet) (out []ReportWorkload, err error) {
+	// Construct a pipeline from the steps; we'll slice it later
+	base, err := Pipeline(pr, steps)
+	if err != nil {
+		return
+	}
+
+	// Prepare workloads based on the provided definitions
+	//
+	// - If consecutive definitions point to the same source with the same name
+	//   consider them to fall under the same workload (the link step)
+	// - else, one def per workload
+	auxDefs := make(types.ReportFrameDefinitionSet, 0)
+	var auxOut ReportWorkload
+	for i, def := range defs {
+		if i == 0 {
+			auxDefs = append(auxDefs, def)
+			continue
+		}
+
+		// This is for the link step
+		if def.Name == defs[i-1].Name && def.Source == defs[i-1].Source {
+			auxDefs = append(auxDefs, def)
+			continue
+		}
+
+		// This is for the rest
+		auxOut, err = makeWorkload(base, auxDefs)
+		if err != nil {
+			return
+		}
+		out = append(out, auxOut)
+		auxDefs = make(types.ReportFrameDefinitionSet, 0)
+		auxDefs = append(auxDefs, def)
+	}
+
+	// Handle the ones (potentially) not covered by the above loop
+	if len(auxDefs) > 0 {
+		auxOut, err = makeWorkload(base, auxDefs)
+		if err != nil {
+			return
+		}
+		out = append(out, auxOut)
+	}
+
+	return
+}
+
+// Frames returns a set of frames for the given workload & iterator combo
+func Frames(ctx context.Context, iter dal.Iterator, workload ReportWorkload) (ff []*types.ReportFrame, err error) {
+	// Preprocessing on the workload's frame definitions; assure all columns/metdata are there
+	// to avoid nonesense later down the line
+	updateDefAttrs(workload)
+
+	// @todo perhaps need to change; for now only this scenario introduces multiple
+	//       frame defs per workload
+	if len(workload.FrameDefs) > 1 {
+		return framifyLinkIter(ctx, iter, workload)
+	}
+
+	return framifyIter(ctx, iter, workload)
+}
+
+// framifyLinkIter is a handler dedicated for the link step due to it's unique output
+func framifyLinkIter(ctx context.Context, iter dal.Iterator, workload ReportWorkload) (ff []*types.ReportFrame, err error) {
+	defs := workload.FrameDefs
+	// @note this will only be called for the link step so it can freely panic if violated
+	r := workload.Pipeline[0].(*dal.Link)
+
+	// Unpack frame definitions for the link
+	defLeft, defRight := unpackLinkDefs(defs, workload.Pipeline)
+
+	// Init vars to keep track of the progress
+	// @note true is left, false is right
+	counters := make(map[bool]uint)
+
+	builders := make(map[bool]*reportFrameBuilder)
+	builders[true] = newReportFrameBuilder(defLeft)
+	builders[false] = newReportFrameBuilder(defRight)
+	builders[false].withRefs(r.On.Right)
+
+	limits := make(map[bool]uint)
+	if defLeft.Paging != nil {
+		limits[true] = defLeft.Paging.Limit
+	}
+	if defRight.Paging != nil {
+		limits[false] = defRight.Paging.Limit
+	}
+
+	// Helper to determine if we need a next cursor
+	nextCursor := false
+
+	// Helpers for reading iterators
+	var (
+		ref    string
+		row    = &dal.Row{}
+		doingF = false
+	)
+
+	for iter.Next(ctx) {
+		if limits[true] > 0 && counters[true] >= limits[true] {
+			nextCursor = true
+			break
+		}
+		row.Reset()
+
+		_ = iter.Scan(row)
+
+		// Determine ref and which vars to use
+		aux, _ := row.GetValue(dal.LinkRefIdent, 0)
+		ref = cast.ToString(aux)
+		if ref == "" {
+			ref = defLeft.Ref
+		}
+		left := ref == defLeft.Ref
+
+		// When needed, flush the finished frames to the output
+		if left && doingF {
+			ff = append(ff, builders[false].done())
+			doingF = false
+		} else if !left {
+			doingF = true
+		}
+
+		builders[left].addRow(row)
+		counters[left]++
+	}
+	if err = iter.Err(); err != nil {
+		return
+	}
+
+	// If the loop ended before the limit cut it off, we need to finish the
+	// last right frame as it wasn't yet in the above loop
+	if !nextCursor {
+		if doingF {
+			ff = append(ff, builders[false].done())
+		}
+	}
+
+	// Apply paging cursor to the left frame
+	// @todo consider applying them to the right as well, for now, no
+	if nextCursor {
+		if builders[true].frame.Paging == nil {
+			builders[true].frame.Paging = &filter.Paging{}
+		}
+		builders[true].frame.Paging.NextPage, err = iter.ForwardCursor(row)
+		if err != nil {
+			return
+		}
+	}
+
+	// Complete the output with the left frame
+	// @note the left frame goes to the start and the right frames are in the same order
+	//       as the related rows from the left frame.
+	return append([]*types.ReportFrame{builders[true].done()}, ff...), nil
+}
+
+// unpackLinkDefs returns the left and the right frame definition disregarding
+// the order of the definitions in the input
+func unpackLinkDefs(defs types.ReportFrameDefinitionSet, pp dal.Pipeline) (left, right *types.ReportFrameDefinition) {
+	r := pp[0]
+
+	l := r.(*dal.Link)
+
+	find := func(defs types.ReportFrameDefinitionSet, ref string) *types.ReportFrameDefinition {
+		for _, def := range defs {
+			if def.Ref == ref {
+				return def
+			}
+		}
+		return nil
+	}
+
+	return find(defs, l.RelLeft), find(defs, l.RelRight)
+}
+
+// framifyIter is a generic iter to frame handler
+func framifyIter(ctx context.Context, iter dal.Iterator, workload ReportWorkload) (ff []*types.ReportFrame, err error) {
+	defs := workload.FrameDefs
+
+	// @note only the link step takes multiple defs and that one is not covered
+	//       by this function
+	if len(defs) != 1 {
+		panic(fmt.Sprintf("impossible state: expecting one frame definition, got %d", len(defs)))
+	}
+	def := defs[0]
+
+	// Init vars to keep track of the progress
+	limit := uint(0)
+	counter := uint(0)
+	builder := newReportFrameBuilder(def)
+	if def.Paging != nil {
+		limit = def.Paging.Limit
+	}
+
+	// Helper to determine if we need a next cursor
+	nextCursor := false
+
+	// Helpers for reading iterators
+	row := &dal.Row{}
+
+	for iter.Next(ctx) {
+		if limit > 0 && counter >= limit {
+			nextCursor = true
+			break
+		}
+		row.Reset()
+
+		_ = iter.Scan(row)
+		builder.addRow(row)
+		counter++
+	}
+	if err = iter.Err(); err != nil {
+		return
+	}
+
+	// Apply paging cursor to the frame
+	if nextCursor {
+		if builder.frame.Paging == nil {
+			builder.frame.Paging = &filter.Paging{}
+		}
+		builder.frame.Paging.NextPage, err = iter.ForwardCursor(row)
+		if err != nil {
+			return
+		}
+	}
+
+	return append(ff, builder.done()), nil
+}
+
+// Pipeline creates a pipeline from the given steps
+func Pipeline(pr PipelineRunner, steps types.ReportStepSet) (out dal.Pipeline, err error) {
+	out = make(dal.Pipeline, 0, len(steps))
+
+	for _, step := range steps {
+		switch {
+		case step.Load != nil:
+			aux, err := makeStepLoad(pr, *step.Load)
+			if err != nil {
+				return nil, err
+			}
+			out = append(out, aux)
+
+		case step.Join != nil:
+			aux, err := makeStepJoin(*step.Join)
+			if err != nil {
+				return nil, err
+			}
+			out = append(out, aux)
+
+		case step.Link != nil:
+			aux, err := makeStepLink(*step.Link)
+			if err != nil {
+				return nil, err
+			}
+			out = append(out, aux)
+
+		case step.Aggregate != nil:
+			aux, err := makeStepAggregate(*step.Aggregate)
+			if err != nil {
+				return nil, err
+			}
+			out = append(out, aux)
+		}
+	}
+
+	return out, out.LinkSteps()
+}
+
+// DescribePipeline returns a set of frame descriptions based on the given pipeline
+func DescribePipeline(pp dal.Pipeline, sources []string) (out types.FrameDescriptionSet, err error) {
+	var aux types.FrameDescriptionSet
+
+	for _, src := range sources {
+		sub := pp.Slice(src)
+		s := sub[0]
+
+		aux, err = describePipeline(s, src)
+		if err != nil {
+			return
+		}
+
+		out = append(out, aux...)
+	}
+
+	return
+}
+
+func describePipeline(s dal.PipelineStep, src string) (out types.FrameDescriptionSet, err error) {
+	aa := s.Attributes()
+
+	out = make(types.FrameDescriptionSet, len(aa))
+	for i, a := range aa {
+		out[i].Source = src
+		out[i].Columns = mappingToFrameCols(a)
+	}
+
+	// @note this case is only possible for the link step; expand when/if needed
+	if len(out) == 2 {
+		l := s.(*dal.Link)
+		out[0].Ref = l.RelLeft
+		out[1].Ref = l.RelRight
+	}
+
+	return
+}
+
+func makeWorkload(pp dal.Pipeline, defs types.ReportFrameDefinitionSet) (out ReportWorkload, err error) {
+	// We'll trust all of the defs point to the same source; this should be validated
+	// way sooner
+	def := defs[0]
+
+	out.FrameDefs = defs
+	out.Pipeline = pp.Slice(def.Source)
+
+	return
+}
+
+// @todo address with col/attr rework/rethink
+func mappingToFrameCols(mm []dal.AttributeMapping) types.ReportFrameColumnSet {
+	out := make(types.ReportFrameColumnSet, 0, len(mm))
+
+	for _, m := range mm {
+		out = append(out, types.ReportFrameColumn{
+			Name: m.Identifier(),
+			Kind: "String",
+		})
+	}
+
+	return out
+}
+
+// Report step -> DAL step conversion
+
+func makeStepJoin(step types.ReportStepJoin) (out *dal.Join, err error) {
+	out = &dal.Join{
+		Ident:    step.Name,
+		RelLeft:  step.LocalSource,
+		RelRight: step.ForeignSource,
+
+		On: dal.JoinPredicate{
+			Left:  step.LocalColumn,
+			Right: step.ForeignColumn,
+		},
+		Filter: dal.FilterForExpr(step.Filter.Node()),
+	}
+	return
+}
+
+func makeStepLink(step types.ReportStepLink) (out *dal.Link, err error) {
+	out = &dal.Link{
+		Ident:    step.Name,
+		RelLeft:  step.LocalSource,
+		RelRight: step.ForeignSource,
+
+		On: dal.LinkPredicate{
+			Left:  step.LocalColumn,
+			Right: step.ForeignColumn,
+		},
+		Filter: dal.FilterForExpr(step.Filter.Node()),
+	}
+	return
+}
+
+func makeStepAggregate(step types.ReportStepAggregate) (out *dal.Aggregate, err error) {
+	out = &dal.Aggregate{
+		Ident:     step.Name,
+		RelSource: step.Source,
+		Filter:    dal.FilterForExpr(step.Filter.Node()),
+
+		Group:         step.Keys.DalMapping(),
+		OutAttributes: step.Columns.DalMapping(),
+	}
+	return
+}
+
+func makeStepLoad(pr PipelineRunner, step types.ReportStepLoad) (out *dal.Datasource, err error) {
+	mfr, c, err := getModelRef(step)
+	if err != nil {
+		return
+	}
+
+	return &dal.Datasource{
+		Ident: step.Name,
+		Filter: dal.FilterForExpr(step.Filter.Node()).
+			WithConstraints(c),
+		ModelRef:      mfr,
+		OutAttributes: filteredModelAttributes(pr, step, mfr),
+	}, nil
+
+}
+
+// // // // // // // // // // // // // // // // // // // // // // // // //
+// Utilities...
+
+// filteredModelAttributes returns the requested attributes based on the step
+// definition or all attributes if none are specified
+//
+// The function collects the attributes from the DAL model to omit the attribute
+// construction step.
+func filteredModelAttributes(pr PipelineRunner, step types.ReportStepLoad, mfr dal.ModelRef) (out []dal.AttributeMapping) {
+	out = make([]dal.AttributeMapping, 0, 100)
+
+	// All of the attributes
+	fullAttrs, err := getModelAttrs(pr, mfr)
+	if err != nil {
+		return
+	}
+
+	// No filtering, return all
+	if len(step.Columns) == 0 {
+		return attrToMapping(fullAttrs...)
+	}
+
+	// Filter out the ones we don't want
+	reqIndex := make(map[string]bool)
+	for _, col := range step.Columns {
+		reqIndex[col.Name] = true
+	}
+
+	for _, attr := range fullAttrs {
+		if reqIndex[attr.Ident] {
+			out = append(out, attrToMapping(attr)...)
+		}
+	}
+
+	return
+}
+
+func getModelAttrs(pr PipelineRunner, mfr dal.ModelRef) (attrs dal.AttributeSet, err error) {
+	m := pr.FindModel(mfr)
+	if m == nil {
+		return nil, fmt.Errorf("model not found: %v", mfr)
+	}
+
+	return m.Attributes, nil
+}
+
+func attrToMapping(aa ...*dal.Attribute) (out []dal.AttributeMapping) {
+	for _, a := range aa {
+		out = append(out, dal.SimpleAttr{
+			Ident: a.Ident,
+			Src:   a.Ident,
+			Props: dal.MapProperties{
+				Type:      a.Type,
+				Nullable:  a.Type.IsNullable(),
+				IsPrimary: a.PrimaryKey,
+			},
+		})
+	}
+	return
+}
+
+// @todo support for ns/mod by handle
+func getModelRef(step types.ReportStepLoad) (out dal.ModelRef, constraints map[string][]any, err error) {
+	var (
+		moduleID, connectionID, namespaceID uint64
+		aux                                 any
+		ok                                  bool
+	)
+
+	constraints = make(map[string][]any)
+
+	if aux, ok = step.Definition["moduleID"]; ok {
+		moduleID = cast.ToUint64(aux)
+	} else if aux, ok = step.Definition["module"]; ok {
+		moduleID = cast.ToUint64(aux)
+	} else {
+		err = fmt.Errorf("step definition is missing moduleID")
+		return
+	}
+
+	if aux, ok = step.Definition["namespaceID"]; ok {
+		namespaceID = cast.ToUint64(aux)
+	} else if aux, ok = step.Definition["module"]; ok {
+		namespaceID = cast.ToUint64(aux)
+	} else {
+		err = fmt.Errorf("step definition is missing namespaceID")
+		return
+	}
+
+	if aux, ok = step.Definition["connectionID"]; ok {
+		connectionID = cast.ToUint64(aux)
+	} else if aux, ok = step.Definition["connection"]; ok {
+		connectionID = cast.ToUint64(aux)
+	}
+
+	constraints["moduleID"] = []any{moduleID}
+	constraints["namespaceID"] = []any{namespaceID}
+
+	return dal.ModelRef{ConnectionID: connectionID, ResourceID: moduleID}, constraints, nil
+}
+
+func updateDefAttrs(workload ReportWorkload) {
+	ppAttrs := workload.Pipeline[0].Attributes()
+	for i, def := range workload.FrameDefs {
+		if len(def.Columns) > 0 {
+			continue
+		}
+
+		def.Columns = mappingToFrameCols(ppAttrs[i])
+	}
+}

system/rest.yaml

@@ -1610,7 +1610,6 @@ endpoints:
   authentication: []
   imports:
     - github.com/cortezaproject/corteza-server/pkg/label
-    - github.com/cortezaproject/corteza-server/pkg/report
     - github.com/cortezaproject/corteza-server/system/types
   apis:
   - name: list
@@ -1705,9 +1704,9 @@ endpoints:
     path: "/describe"
     parameters:
       post:
-      - { name: sources,  type: "types.ReportDataSourceSet",  title: Report steps definition }
-      - { name: steps,    type: "report.StepDefinitionSet",   title: Report steps definition  }
-      - { name: describe, type: "[]string",                   title: The source descriptions to generate }
+      - { name: sources,  type: "types.ReportDataSourceSet", title: Report steps definition }
+      - { name: steps,    type: "types.ReportStepSet",       title: Report steps definition  }
+      - { name: describe, type: "[]string",                  title: The source descriptions to generate }
 
   - name: run
     method: POST
@@ -1720,7 +1719,7 @@ endpoints:
         required: true
         title: Report ID
       post:
-      - { name: frames,       type: "report.FrameDefinitionSet",  title: Report data frame definitions }
+      - { name: frames,       type: "types.ReportFrameDefinitionSet",  title: Report data frame definitions }
 - title: Statistics
   entrypoint: stats
   path: "/stats"

system/rest/report.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/cortezaproject/corteza-server/pkg/api"
 	"github.com/cortezaproject/corteza-server/pkg/filter"
-	"github.com/cortezaproject/corteza-server/pkg/report"
 	"github.com/cortezaproject/corteza-server/system/rest/request"
 	"github.com/cortezaproject/corteza-server/system/service"
 	"github.com/cortezaproject/corteza-server/system/types"
@@ -27,8 +26,8 @@ type (
 		Update(ctx context.Context, upd *types.Report) (app *types.Report, err error)
 		Delete(ctx context.Context, ID uint64) (err error)
 		Undelete(ctx context.Context, ID uint64) (err error)
-		Run(ctx context.Context, ID uint64, dd report.FrameDefinitionSet) (rr []*report.Frame, err error)
-		DescribeFresh(ctx context.Context, src types.ReportDataSourceSet, st report.StepDefinitionSet, sources ...string) (out report.FrameDescriptionSet, err error)
+		Run(ctx context.Context, ID uint64, dd types.ReportFrameDefinitionSet) (rr []*types.ReportFrame, err error)
+		Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out types.FrameDescriptionSet, err error)
 	}
 
 	reportAccessController interface {
@@ -56,7 +55,7 @@ type (
 	}
 
 	reportFramePayload struct {
-		Frames []*report.Frame `json:"frames"`
+		Frames []*types.ReportFrame `json:"frames"`
 	}
 )
 
@@ -138,7 +137,7 @@ func (ctrl *Report) Undelete(ctx context.Context, r *request.ReportUndelete) (in
 }
 
 func (ctrl *Report) Describe(ctx context.Context, r *request.ReportDescribe) (interface{}, error) {
-	return ctrl.report.DescribeFresh(ctx, r.Sources, r.Steps, r.Describe...)
+	return ctrl.report.Describe(ctx, r.Sources, r.Steps, r.Describe...)
 }
 
 func (ctrl *Report) Run(ctx context.Context, r *request.ReportRun) (interface{}, error) {
@@ -177,7 +176,7 @@ func (ctrl Report) makeFilterPayload(ctx context.Context, nn types.ReportSet, f
 	return msp, nil
 }
 
-func (ctrl Report) makeReportFramePayload(ctx context.Context, ff []*report.Frame, err error) (*reportFramePayload, error) {
+func (ctrl Report) makeReportFramePayload(ctx context.Context, ff []*types.ReportFrame, err error) (*reportFramePayload, error) {
 	if err != nil || len(ff) == 0 {
 		return nil, err
 	}

system/rest/request/report.go

@@ -13,7 +13,6 @@ import (
 	"fmt"
 	"github.com/cortezaproject/corteza-server/pkg/label"
 	"github.com/cortezaproject/corteza-server/pkg/payload"
-	"github.com/cortezaproject/corteza-server/pkg/report"
 	"github.com/cortezaproject/corteza-server/system/types"
 	"github.com/go-chi/chi/v5"
 	"io"
@@ -167,7 +166,7 @@ type (
 		// Steps POST parameter
 		//
 		// Report steps definition
-		Steps report.StepDefinitionSet
+		Steps types.ReportStepSet
 
 		// Describe POST parameter
 		//
@@ -184,7 +183,7 @@ type (
 		// Frames POST parameter
 		//
 		// Report data frame definitions
-		Frames report.FrameDefinitionSet
+		Frames types.ReportFrameDefinitionSet
 	}
 )
 
@@ -759,7 +758,7 @@ func (r ReportDescribe) GetSources() types.ReportDataSourceSet {
 }
 
 // Auditable returns all auditable/loggable parameters
-func (r ReportDescribe) GetSteps() report.StepDefinitionSet {
+func (r ReportDescribe) GetSteps() types.ReportStepSet {
 	return r.Steps
 }
 
@@ -807,7 +806,7 @@ func (r *ReportDescribe) Fill(req *http.Request) (err error) {
 		//}
 
 		//if val, ok := req.Form["steps[]"]; ok && len(val) > 0  {
-		//    r.Steps, err = report.StepDefinitionSet(val), nil
+		//    r.Steps, err = types.ReportStepSet(val), nil
 		//    if err != nil {
 		//        return err
 		//    }
@@ -843,7 +842,7 @@ func (r ReportRun) GetReportID() uint64 {
 }
 
 // Auditable returns all auditable/loggable parameters
-func (r ReportRun) GetFrames() report.FrameDefinitionSet {
+func (r ReportRun) GetFrames() types.ReportFrameDefinitionSet {
 	return r.Frames
 }
 
@@ -879,7 +878,7 @@ func (r *ReportRun) Fill(req *http.Request) (err error) {
 		// POST params
 
 		//if val, ok := req.Form["frames[]"]; ok && len(val) > 0  {
-		//    r.Frames, err = report.FrameDefinitionSet(val), nil
+		//    r.Frames, err = types.ReportFrameDefinitionSet(val), nil
 		//    if err != nil {
 		//        return err
 		//    }

system/service/report.go

@@ -2,15 +2,15 @@ package service
 
 import (
 	"context"
-	"fmt"
-	"github.com/cortezaproject/corteza-server/pkg/errors"
 	"strconv"
 
+	"github.com/cortezaproject/corteza-server/pkg/dal"
+	"github.com/cortezaproject/corteza-server/pkg/errors"
+	"github.com/cortezaproject/corteza-server/pkg/reportutils"
+
 	"github.com/cortezaproject/corteza-server/pkg/actionlog"
-	"github.com/cortezaproject/corteza-server/pkg/expr"
 	"github.com/cortezaproject/corteza-server/pkg/filter"
 	"github.com/cortezaproject/corteza-server/pkg/label"
-	rep "github.com/cortezaproject/corteza-server/pkg/report"
 	"github.com/cortezaproject/corteza-server/store"
 	"github.com/cortezaproject/corteza-server/system/types"
 	"github.com/modern-go/reflect2"
@@ -25,6 +25,8 @@ type (
 		store     store.Storer
 
 		users UserService
+
+		pipelineRunner reportutils.PipelineRunner
 	}
 
 	reportAccessController interface {
@@ -38,7 +40,7 @@ type (
 )
 
 var (
-	reporters = make(map[string]rep.DatasourceProvider)
+	reporters = make(map[string]any)
 )
 
 // Report is a default report service initializer
@@ -50,10 +52,12 @@ func Report(s store.Storer, ac reportAccessController, al actionlog.Recorder, eb
 		eventbus:  eb,
 
 		users: DefaultUser,
+
+		pipelineRunner: dal.Service(),
 	}
 }
 
-func (svc *report) RegisterReporter(key string, r rep.DatasourceProvider) {
+func (svc *report) RegisterReporter(key string, r any) {
 	reporters[key] = r
 }
 
@@ -291,12 +295,8 @@ func (svc *report) Undelete(ctx context.Context, ID uint64) (err error) {
 }
 
 // actionlog?
-func (svc *report) DescribeFresh(ctx context.Context, src types.ReportDataSourceSet, st rep.StepDefinitionSet, sources ...string) (out rep.FrameDescriptionSet, err error) {
-	// var (
-	// 	aaProps = &reportActionProps{}
-	// )
-
-	out = make(rep.FrameDescriptionSet, 0, len(sources)*2)
+func (svc *report) Describe(ctx context.Context, src types.ReportDataSourceSet, st types.ReportStepSet, sources ...string) (out types.FrameDescriptionSet, err error) {
+	out = make(types.FrameDescriptionSet, 0, len(sources)*2)
 
 	err = func() (err error) {
 		if !svc.ac.CanCreateReport(ctx) {
@@ -306,124 +306,80 @@ func (svc *report) DescribeFresh(ctx context.Context, src types.ReportDataSource
 		ss := src.ModelSteps()
 		ss = append(ss, st...)
 
-		// Model the report
-		model, err := rep.Model(ctx, reporters, ss...)
+		pp, err := reportutils.Pipeline(svc.pipelineRunner, ss)
 		if err != nil {
 			return
 		}
 
-		err = model.Run(ctx)
+		err = pp.LinkSteps()
 		if err != nil {
 			return
 		}
 
-		var auxOut rep.FrameDescriptionSet
-		for _, s := range sources {
-			auxOut, err = model.Describe(ctx, s)
-			if err != nil {
-				return err
-			}
-
-			out = append(out, auxOut...)
+		// Dryrun the pipeline to go over all init steps
+		err = svc.pipelineRunner.Dryrun(ctx, pp)
+		if err != nil {
+			return
 		}
 
-		return nil
+		out, err = reportutils.DescribePipeline(pp, sources)
+		return err
 	}()
 
 	return out, err
 }
 
-func (svc *report) Run(ctx context.Context, reportID uint64, dd rep.FrameDefinitionSet) (out []*rep.Frame, err error) {
+func (svc *report) Run(ctx context.Context, reportID uint64, dd types.ReportFrameDefinitionSet) (out []*types.ReportFrame, err error) {
 	var (
 		aaProps = &reportActionProps{}
+
+		iter dal.Iterator
+		ff   []*types.ReportFrame
 	)
-	out = make([]*rep.Frame, 0, 4)
+	out = make([]*types.ReportFrame, 0, 4)
 
 	err = func() (err error) {
-		// @todo evt bus?
-		// if err = svc.eventbus.WaitFor(ctx, event.ReportBeforeUpdate(upd, report)); err != nil {
-		// 	return
-		// }
-
+		// Load the report
 		r, err := loadReport(ctx, svc.store, reportID)
 		if err != nil {
 			return err
 		}
 
-		// - ac
+		// Access control
 		if !svc.ac.CanRunReport(ctx, r) {
 			return ReportErrNotAllowedToRun()
 		}
 
+		// Get all of the steps
 		ss := r.Sources.ModelSteps()
 		ss = append(ss, r.Blocks.ModelSteps()...)
 
-		if err = ss.Validate(); err != nil {
-			return ReportErrInvalidConfiguration().Wrap(err)
-		}
-
-		for _, d := range dd {
-			if err = d.Validate(); err != nil {
-				return ReportErrInvalidConfiguration().Wrap(err)
-			}
-		}
-
-		// Model the report
-		model, err := rep.Model(ctx, reporters, ss...)
+		// Prepare required pipelines
+		workloads, err := reportutils.Workloads(svc.pipelineRunner, ss, dd)
 		if err != nil {
 			return
 		}
 
-		err = model.Run(ctx)
-		if err != nil {
-			return
-		}
-
-		auxdd := make([]*rep.FrameDefinition, 0, len(dd))
-		for i, d := range dd {
-			// first one; nothing special needed
-			if i == 0 {
-				auxdd = append(auxdd, d)
-				continue
-			}
-
-			stp := model.GetStep(d.Source)
-			if stp == nil {
-				return fmt.Errorf("unknown source: %s", d.Source)
-			}
-
-			// if the current source matches the prev. source, and they both define references,
-			// they fall into the same chunk.
-			if stp.Def().Join != nil && (d.Source == dd[i-1].Source) && (d.Ref != "" && dd[i-1].Ref != "") && (d.Name == dd[i-1].Name) {
-				auxdd = append(auxdd, d)
-				continue
-			}
-
-			// if the current one doesn't fall into the current chunk, process
-			// the chunk and reset it
-			ff, err := model.Load(ctx, auxdd...)
+		// Run the pipelines and produce report frames
+		for _, workload := range workloads {
+			iter, err = svc.pipelineRunner.Run(ctx, workload.Pipeline)
 			if err != nil {
-				return err
+				return
 			}
-			out = append(out, ff...)
 
-			auxdd = make([]*rep.FrameDefinition, 0, len(dd))
-			auxdd = append(auxdd, d)
-		}
-
-		if len(auxdd) > 0 {
-			ff, err := model.Load(ctx, auxdd...)
+			ff, err = reportutils.Frames(ctx, iter, workload)
 			if err != nil {
-				return err
+				return
 			}
 
-			ff, err = svc.enhance(ctx, ff)
+			err = svc.enhance(ctx, ff)
+			if err != nil {
+				return
+			}
 
 			out = append(out, ff...)
 		}
 
-		// _ = svc.eventbus.WaitFor(ctx, event.ReportAfterUpdate(upd, report))
-		// return nil
 		return nil
 	}()
 
@@ -434,7 +390,7 @@ func (svc *report) Run(ctx context.Context, reportID uint64, dd rep.FrameDefinit
 // @todo extend core implementation to support such operatons
 //
 // - userID is replaced by the user name || username || email || handle || userID
-func (svc *report) enhance(ctx context.Context, ff []*rep.Frame) (_ []*rep.Frame, err error) {
+func (svc *report) enhance(ctx context.Context, ff []*types.ReportFrame) (err error) {
 	// Preload sys users
 	uIndex := make(map[uint64]*types.User)
 	uu, uf, err := svc.users.Find(ctx, types.UserFilter{Paging: filter.Paging{Limit: 1024}})
@@ -446,6 +402,7 @@ func (svc *report) enhance(ctx context.Context, ff []*rep.Frame) (_ []*rep.Frame
 		uIndex[uu[i].ID] = uu[i]
 	}
 
+	var uID uint64
 	for _, f := range ff {
 		userCols := make([]int, 0, len(f.Columns))
 		for i, c := range f.Columns {
@@ -461,43 +418,42 @@ func (svc *report) enhance(ctx context.Context, ff []*rep.Frame) (_ []*rep.Frame
 				if reflect2.IsNil(col) {
 					continue
 				}
-				switch col.Type() {
-				case "ID":
-					uID := col.Get().(uint64)
-					user, ok := uIndex[uID]
-					if !ok && hasMore {
-						user, err = svc.users.FindByID(ctx, uID)
-						if err != nil && err != store.ErrNotFound {
-							return
-						}
+				uID, err = cast.ToUint64E(col)
+				if err != nil {
+					continue
+				}
+
+				user, ok := uIndex[uID]
+				if !ok && hasMore {
+					user, err = svc.users.FindByID(ctx, uID)
+					if err != nil && err != store.ErrNotFound {
+						return
 					}
+				}
 
-					if user == nil {
-						continue
-					} else if _, ok := uIndex[uID]; !ok {
-						uIndex[uID] = user
-					}
+				if user == nil {
+					continue
+				} else if _, ok := uIndex[uID]; !ok {
+					uIndex[uID] = user
+				}
 
-					if usr, ok := uIndex[uID]; ok {
-						label := strconv.FormatUint(uID, 10)
-						if usr.Name != "" {
-							label = usr.Name
-						} else if usr.Username != "" {
-							label = usr.Username
-						} else if usr.Email != "" {
-							label = usr.Email
-						} else if usr.Handle != "" {
-							label = usr.Handle
-						}
-
-						r[ci], _ = expr.NewString(label)
+				if usr, ok := uIndex[uID]; ok {
+					r[ci] = strconv.FormatUint(uID, 10)
+					if usr.Name != "" {
+						r[ci] = usr.Name
+					} else if usr.Username != "" {
+						r[ci] = usr.Username
+					} else if usr.Email != "" {
+						r[ci] = usr.Email
+					} else if usr.Handle != "" {
+						r[ci] = usr.Handle
 					}
 				}
 			}
 		}
 	}
 
-	return ff, err
+	return err
 }
 
 func (svc *report) setIDs(r *types.Report) *types.Report {

system/types/report.go

@@ -239,6 +239,13 @@ func (vv ReportDataSourceSet) Value() (driver.Value, error) { return json.Marsha
 func (vv *ReportScenarioSet) Scan(src any) error          { return sql.ParseJSON(src, vv) }
 func (vv ReportScenarioSet) Value() (driver.Value, error) { return json.Marshal(vv) }
 
+func (f *qlExprWrap) Node() *ql.ASTNode {
+	if f == nil {
+		return nil
+	}
+	return f.ASTNode
+}
+
 func (f *qlExprWrap) UnmarshalJSON(data []byte) (err error) {
 	var aux interface{}
 	if err = json.Unmarshal(data, &aux); err != nil {