Base ComposeRecord Datasource definition

compose/service/record.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cortezaproject/corteza-server/pkg/errors"
 	"github.com/cortezaproject/corteza-server/pkg/eventbus"
 	"github.com/cortezaproject/corteza-server/pkg/label"
+	"github.com/cortezaproject/corteza-server/pkg/report"
 	"github.com/cortezaproject/corteza-server/store"
 )
 
@@ -85,6 +86,8 @@ 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)
@@ -1373,7 +1376,6 @@ func (svc record) Iterator(ctx context.Context, f types.RecordFilter, fn eventbu
 	}()
 
 	return svc.recordAction(ctx, aProps, RecordActionIteratorInvoked, err)
-
 }
 
 func ComposeRecordFilterChecker(ctx context.Context, ac recordAccessController, m *types.Module) func(*types.Record) (bool, error) {

compose/service/record_datasource.go

@@ -0,0 +1,123 @@
+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")
+	}
+
+	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 := svc.store.LookupComposeModuleByID(ctx, moduleID)
+	if err != nil {
+		return nil, err
+	}
+	mod.Fields, _, err = svc.store.SearchComposeModuleFields(ctx, 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.Name = "id"
+		c.Label = "Record ID"
+		cols = append(cols, c)
+
+		for _, f := range mod.Fields {
+			k := f.Kind
+			c = report.MakeColumnOfKind(k)
+			c.Name = f.Name
+			c.Label = f.Label
+			cols = append(cols, c)
+		}
+
+		// Sys fields
+		c = report.MakeColumnOfKind("Date")
+		c.Name = "createdAt"
+		c.Label = "Created at"
+		cols = append(cols, c)
+
+		c = report.MakeColumnOfKind("User")
+		c.Name = "createdBy"
+		c.Label = "Created by"
+		cols = append(cols, c)
+
+		c = report.MakeColumnOfKind("Date")
+		c.Name = "updatedAt"
+		c.Label = "Updated at"
+		cols = append(cols, c)
+
+		c = report.MakeColumnOfKind("User")
+		c.Name = "updatedBy"
+		c.Label = "Updated by"
+		cols = append(cols, c)
+
+		c = report.MakeColumnOfKind("Date")
+		c.Name = "deletedAt"
+		c.Label = "Deleted at"
+		cols = append(cols, c)
+
+		c = report.MakeColumnOfKind("User")
+		c.Name = "deletedBy"
+		c.Label = "Deleted by"
+		cols = append(cols, c)
+
+		ld.Columns = cols
+	}
+
+	return svc.store.ComposeRecordDatasource(ctx, mod, ld)
+}

compose/service/service.go

@@ -3,6 +3,9 @@ package service
 import (
 	"context"
 	"fmt"
+	"strconv"
+	"time"
+
 	automationService "github.com/cortezaproject/corteza-server/automation/service"
 	"github.com/cortezaproject/corteza-server/compose/automation"
 	"github.com/cortezaproject/corteza-server/compose/types"
@@ -18,9 +21,8 @@ import (
 	"github.com/cortezaproject/corteza-server/pkg/objstore/plain"
 	"github.com/cortezaproject/corteza-server/pkg/options"
 	"github.com/cortezaproject/corteza-server/store"
+	systemService "github.com/cortezaproject/corteza-server/system/service"
 	"go.uber.org/zap"
-	"strconv"
-	"time"
 )
 
 type (
@@ -172,6 +174,10 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, c Config)
 		DefaultNamespace,
 	)
 
+	// Register reporters
+	// @todo additional datasource providers; generate?
+	systemService.DefaultReport.RegisterReporter("composeRecords", DefaultRecord)
+
 	return nil
 }
 

store/compose_records.gen.go

@@ -10,7 +10,9 @@ package store
 
 import (
 	"context"
+
 	"github.com/cortezaproject/corteza-server/compose/types"
+	"github.com/cortezaproject/corteza-server/pkg/report"
 )
 
 type (
@@ -36,6 +38,9 @@ type (
 
 		// PartialComposeRecordValueUpdate (custom function)
 		PartialComposeRecordValueUpdate(ctx context.Context, _mod *types.Module, _values ...*types.RecordValue) error
+
+		// @todo !!!
+		ComposeRecordDatasource(ctx context.Context, _mod *types.Module, _ld *report.LoadStepDefinition) (report.Datasource, error)
 	}
 )
 
@@ -90,3 +95,7 @@ func ComposeRecordReport(ctx context.Context, s ComposeRecords, _mod *types.Modu
 func PartialComposeRecordValueUpdate(ctx context.Context, s ComposeRecords, _mod *types.Module, _values ...*types.RecordValue) error {
 	return s.PartialComposeRecordValueUpdate(ctx, _mod, _values...)
 }
+
+func ComposeRecordDatasource(ctx context.Context, s ComposeRecords, _mod *types.Module, _ld *report.LoadStepDefinition) (report.Datasource, error) {
+	return s.ComposeRecordDatasource(ctx, _mod, _ld)
+}

store/rdbms/compose_record_datasource.go

@@ -0,0 +1,439 @@
+package rdbms
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+
+	"github.com/Masterminds/squirrel"
+	"github.com/cortezaproject/corteza-server/compose/types"
+	"github.com/cortezaproject/corteza-server/pkg/ql"
+	"github.com/cortezaproject/corteza-server/pkg/report"
+	"github.com/cortezaproject/corteza-server/pkg/slice"
+	"github.com/jmoiron/sqlx"
+)
+
+type (
+	recordDatasource struct {
+		name   string
+		module *types.Module
+
+		// @todo use these
+		supportedAggregationFunctions map[string]bool
+		supportedFilterFunctions      map[string]bool
+
+		store *Store
+		rows  *sql.Rows
+
+		baseFilter *report.RowDefinition
+
+		cols         report.FrameColumnSet
+		qBuilder     squirrel.SelectBuilder
+		nestLevel    int
+		levelColumns map[string]bool
+	}
+)
+
+func ComposeRecordDatasourceBuilder(s *Store, module *types.Module, ld *report.LoadStepDefinition) (report.Datasource, error) {
+	var err error
+
+	r := &recordDatasource{
+		name:         ld.Name,
+		module:       module,
+		store:        s,
+		cols:         ld.Columns,
+		levelColumns: make(map[string]bool),
+
+		supportedAggregationFunctions: slice.ToStringBoolMap([]string{
+			"COUNT",
+			"SUM",
+			"MAX",
+			"MIN",
+			"AVG",
+		}),
+
+		supportedFilterFunctions: slice.ToStringBoolMap([]string{
+			"CONCAT",
+			"QUARTER",
+			"YEAR",
+			"DATE",
+			"NOW",
+			"DATE_ADD",
+			"DATE_SUB",
+			"DATE_FORMAT",
+		}),
+	}
+
+	r.qBuilder, err = r.baseQuery(ld.Rows)
+
+	return r, err
+}
+
+func (r *recordDatasource) Name() string {
+	if r.name != "" {
+		return r.name
+	}
+
+	if r.module == nil {
+		return ""
+	}
+
+	if r.module.Handle == "" {
+		return r.module.Name
+	}
+	return r.module.Handle
+}
+
+// @todo add Transform
+// @todo try to make Group and Transform use the base query
+
+func (r *recordDatasource) Group(d report.GroupDefinition, name string) (bool, error) {
+	defer func() {
+		r.nestLevel++
+		r.name = name
+	}()
+
+	cls := r.levelColumns
+	r.levelColumns = make(map[string]bool)
+
+	gCols := make(report.FrameColumnSet, 0, 10)
+
+	q := squirrel.Select()
+
+	// @todo allow some transformation functions within the agg. functions
+	parser := ql.NewParser()
+	parser.OnFunction = r.stdAggregationHandler
+	parser.OnIdent = func(i ql.Ident) (ql.Ident, error) {
+		if !cls[i.Value] {
+			return i, fmt.Errorf("column %s does not exist on level %d", i.Value, r.nestLevel)
+		}
+
+		i.Value = fmt.Sprintf("l%d.%s", r.nestLevel, i.Value)
+		return i, nil
+	}
+
+	for _, g := range d.Groups {
+		e, err := parser.ParseExpression(g.Expr)
+		if err != nil {
+			return false, err
+		}
+
+		// @todo imply based on context
+		c := report.MakeColumnOfKind(g.Kind)
+		c.Name = g.Name
+		gCols = append(gCols, c)
+
+		r.levelColumns[g.Name] = true
+		q = q.Column(fmt.Sprintf("(%s) as `%s`", e.String(), g.Name)).
+			GroupBy(e.String())
+	}
+
+	for _, c := range d.Columns {
+		for alias, op := range c {
+			o := op.GetOp()
+			e, err := parser.ParseExpression(fmt.Sprintf("%s(%s)", o, op[o]))
+			if err != nil {
+				return false, err
+			}
+
+			// @todo imply based on context
+			c := report.MakeColumnOfKind("Number")
+			c.Name = alias
+			gCols = append(gCols, c)
+
+			r.levelColumns[alias] = true
+			q = q.
+				Column(fmt.Sprintf("%s as `%s`", e.String(), alias))
+		}
+	}
+
+	if d.Rows != nil {
+		// @todo validate groupping functions
+		hh, err := r.rowFilterToString("", gCols, d.Rows)
+		if err != nil {
+			return false, err
+		}
+		q = q.Having(hh)
+	}
+
+	r.cols = gCols
+	r.qBuilder = q.FromSelect(r.qBuilder, fmt.Sprintf("l%d", r.nestLevel))
+	return true, nil
+}
+
+func (r *recordDatasource) Load(ctx context.Context, dd ...*report.FrameDefinition) (l report.Loader, c report.Closer, err error) {
+	def := dd[0]
+
+	// assure columns
+	// - undefined columns = all columns
+	if len(def.Columns) == 0 {
+		def.Columns = r.cols
+	} else {
+		// - make sure they exist
+		cc := make(report.FrameColumnSet, len(def.Columns))
+		for i, c := range def.Columns {
+			ci := r.cols.Find(c.Name)
+			if ci == -1 {
+				return nil, nil, fmt.Errorf("column not found: %s", c.Name)
+			}
+			cc[i] = r.cols[ci]
+		}
+		def.Columns = cc
+	}
+
+	q := r.qBuilder
+
+	// when filtering/sorting, wrap the base query in a sub-select, so we don't need to
+	// worry about exact column names.
+	//
+	// @todo flatten the query
+	if def.Rows != nil || def.Sorting != nil {
+		wrap := squirrel.Select("*").FromSelect(q, "w_base")
+
+		// additional filtering
+		if def.Rows != nil {
+			f, err := r.rowFilterToString("", r.cols, def.Rows)
+			if err != nil {
+				return nil, nil, err
+			}
+			wrap = wrap.Where(f)
+		}
+
+		// additional sorting
+		if len(def.Sorting) > 0 {
+			ss := make([]string, len(def.Sorting))
+			for i, c := range def.Sorting {
+				ci := r.cols.Find(c.Column)
+				if ci == -1 {
+					return nil, nil, fmt.Errorf("sort column not resolved: %s", c.Column)
+				}
+
+				_, _, typeCast, err := r.store.config.CastModuleFieldToColumnType(r.cols[ci], c.Column)
+				if err != nil {
+					return nil, nil, err
+				}
+
+				ss[i] = r.store.config.SqlSortHandler(fmt.Sprintf(typeCast, c.Column), c.Descending)
+			}
+
+			wrap = wrap.OrderBy(ss...)
+		}
+
+		q = wrap
+	}
+
+	r.rows, err = r.store.Query(ctx, q)
+	if err != nil {
+		return nil, nil, fmt.Errorf("cannot execute query: %w", err)
+	}
+
+	return func(cap int) ([]*report.Frame, error) {
+			f := &report.Frame{
+				Name:   def.Name,
+				Source: def.Source,
+				Ref:    def.Ref,
+			}
+
+			// fetch & convert the data
+			i := 0
+			// @todo make it in place
+			f.Columns = def.Columns
+			f.Rows = make(report.FrameRowSet, 0, cap)
+
+			for r.rows.Next() {
+				i++
+
+				err = r.Cast(r.rows, f)
+				if err != nil {
+					return nil, err
+				}
+
+				if i >= cap {
+					out := []*report.Frame{f}
+					f = &report.Frame{}
+					i = 0
+					return out, nil
+				}
+			}
+
+			if i > 0 {
+				return []*report.Frame{f}, nil
+			}
+			return nil, nil
+		}, func() {
+			if r.rows == nil {
+				return
+			}
+			r.rows.Close()
+		}, nil
+}
+
+// @todo handle those rv_ prefixes; for now omitted
+func (r *recordDatasource) baseQuery(f *report.RowDefinition) (sqb squirrel.SelectBuilder, err error) {
+	var (
+		joinTpl = "compose_record_value AS %s ON (%s.record_id = crd.id AND %s.name = '%s' AND %s.deleted_at IS NULL)"
+
+		report = r.store.composeRecordsSelectBuilder().
+			Where("crd.deleted_at IS NULL").
+			Where("crd.module_id = ?", r.module.ID)
+	)
+
+	// Prepare all of the mod columns
+	// @todo make this as small as possible!
+	for _, f := range r.module.Fields {
+		report = report.LeftJoin(strings.ReplaceAll(joinTpl, "%s", f.Name)).
+			Column(f.Name + ".value as " + f.Name)
+
+		r.levelColumns[f.Name] = true
+	}
+
+	if f != nil {
+		// @todo functions and function validation
+		parser := ql.NewParser()
+		parser.OnIdent = func(i ql.Ident) (ql.Ident, error) {
+			if !r.levelColumns[i.Value] {
+				return i, fmt.Errorf("column %s does not exist on level %d", i.Value, r.nestLevel)
+			}
+
+			return i, nil
+		}
+
+		fl, err := r.rowFilterToString("", r.cols, f)
+		if err != nil {
+			return sqb, err
+		}
+		astq, err := parser.ParseExpression(fl)
+		if err != nil {
+			return sqb, err
+		}
+		report = report.Where(astq.String())
+	}
+
+	return report, nil
+}
+
+func (b *recordDatasource) Cast(row sqlx.ColScanner, out *report.Frame) error {
+	var err error
+	aux := make(map[string]interface{})
+	if err = sqlx.MapScan(row, aux); err != nil {
+		return err
+	}
+
+	r := make(report.FrameRow, len(out.Columns))
+
+	k := ""
+	for i, c := range out.Columns {
+		k = "" + c.Name
+		v, ok := aux[k]
+		if !ok {
+			continue
+		}
+
+		// @todo improve json value casting; I couldn't figure it out at the time
+
+		// @todo this doesn't work...
+		// var aux interface{}
+		// err = json.Unmarshal(v.([]byte), &aux)
+		// if err != nil {
+		// 	return err
+		// }
+
+		switch cv := v.(type) {
+		case []byte:
+			c, err := c.Caster(string(cv))
+			if err != nil {
+				return err
+			}
+			r[i] = c
+
+		case uint64, int64:
+			r[i], err = c.Caster(cv)
+			if err != nil {
+				return err
+			}
+		}
+
+	}
+
+	out.Rows = append(out.Rows, r)
+	return nil
+}
+
+// Identifiers should be names of the fields (physical table columns OR json fields, defined in module)
+func (b *recordDatasource) stdAggregationHandler(f ql.Function) (ql.ASTNode, error) {
+	if !b.supportedAggregationFunctions[strings.ToUpper(f.Name)] {
+		return f, fmt.Errorf("unsupported aggregate function %q", f.Name)
+	}
+
+	return b.store.SqlFunctionHandler(f)
+}
+
+func (ds *recordDatasource) rowFilterToString(conjunction string, cc report.FrameColumnSet, def ...*report.RowDefinition) (string, error) {
+	// The fields on the root level of the definition take priority
+	base := ""
+	for _, f := range def {
+		if f == nil {
+			continue
+		}
+
+		if f.Cells != nil {
+			for k, op := range f.Cells {
+				ci := cc.Find(k)
+				if ci == -1 {
+					return "", fmt.Errorf("filtered column not found in the data frame: %s", k)
+				}
+				_, _, typeCast, err := ds.store.config.CastModuleFieldToColumnType(cc[ci], k)
+				if err != nil {
+					return "", err
+				}
+
+				col := fmt.Sprintf(typeCast, k)
+				base += fmt.Sprintf("%s%s%s %s ", col, op.OpToCmp(), fmt.Sprintf(typeCast, op.Value), conjunction)
+			}
+		}
+	}
+	if strings.TrimSpace(base) != "" {
+		base = strings.TrimSuffix(base, " "+conjunction+" ")
+	}
+
+	// Nested AND
+	for _, f := range def {
+		if f == nil {
+			continue
+		}
+
+		if f.And != nil {
+			na, err := ds.rowFilterToString("AND", cc, f.And...)
+			if err != nil {
+				return "", err
+			}
+			if strings.TrimSpace(base) != "" {
+				base += fmt.Sprintf(" %s (%s)", conjunction, na)
+			} else {
+				base = na
+			}
+		}
+	}
+
+	// Nested OR
+	for _, f := range def {
+		if f == nil {
+			continue
+		}
+
+		if f.Or != nil {
+			na, err := ds.rowFilterToString("OR", cc, f.Or...)
+			if err != nil {
+				return "", err
+			}
+			if strings.TrimSpace(base) != "" {
+				base += fmt.Sprintf(" %s (%s)", conjunction, na)
+			} else {
+				base = na
+			}
+		}
+	}
+
+	return strings.TrimSpace(base), nil
+}

store/rdbms/compose_records.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cortezaproject/corteza-server/compose/types"
 	"github.com/cortezaproject/corteza-server/pkg/filter"
 	"github.com/cortezaproject/corteza-server/pkg/ql"
+	"github.com/cortezaproject/corteza-server/pkg/report"
 	"github.com/cortezaproject/corteza-server/store"
 	"github.com/cortezaproject/corteza-server/store/rdbms/builders"
 )
@@ -447,6 +448,10 @@ func (s Store) ComposeRecordReport(ctx context.Context, m *types.Module, metrics
 	return ComposeRecordReportBuilder(&s, m, metrics, dimensions, filter).Run(ctx)
 }
 
+func (s Store) ComposeRecordDatasource(ctx context.Context, m *types.Module, ld *report.LoadStepDefinition) (report.Datasource, error) {
+	return ComposeRecordDatasourceBuilder(&s, m, ld)
+}
+
 func (s Store) convertComposeRecordFilter(m *types.Module, f types.RecordFilter) (query squirrel.SelectBuilder, err error) {
 	if m == nil {
 		err = fmt.Errorf("module not provided")