Report run handling and validation post testing tweaks

pkg/report/model_graph.go

@@ -18,8 +18,8 @@ type (
 func (m *model) buildStepGraph(ss stepSet) (map[string]*modelGraphNode, error) {
 	mp := make(map[string]*modelGraphNode)
 
-	for _, s := range ss {
-		s := s
+	for _, _s := range ss {
+		s := _s
 
 		// make sure that the step is in the graph
 		n, ok := mp[s.Name()]
@@ -74,15 +74,19 @@ func (m *model) datasource(ctx context.Context, def *FrameDefinition) (ds Dataso
 }
 
 func (m *model) validateBranch(n *modelGraphNode) (err error) {
+	if n.step == nil {
+		return fmt.Errorf("step not defined")
+	}
+
 	err = n.step.Validate()
 	if err != nil {
-		return err
+		return fmt.Errorf("%s: %w", n.step.Name(), err)
 	}
 
 	for _, c := range n.cc {
 		err = m.validateBranch(c)
 		if err != nil {
-			return err
+			return fmt.Errorf("%s: %w", n.step.Name(), err)
 		}
 	}
 

pkg/report/step_join.go

@@ -254,6 +254,12 @@ func (d *joinedDataset) Load(ctx context.Context, dd ...*FrameDefinition) (l Loa
 					break
 				}
 
+				// Cut out any local rows that do not have any foreign rows
+				modified = d.validateRows(mainLoader, subLoader, inverted)
+				if modified {
+					continue
+				}
+
 				// . Sort the collected data
 				//
 				// @todo Most of the sorting has already been done by the datasource,
@@ -483,3 +489,38 @@ func (d *joinedDataset) prepareSorting(local, foreign *FrameDefinition) (inverte
 
 	return
 }
+
+func (d *joinedDataset) validateRows(local, foreign *frameBuffer, inverted bool) (modified bool) {
+	keyCol := foreign.localFrames[0].RelColumn
+	cols := local.localFrames[0].Columns
+	keyColIx := cols.Find(keyCol)
+
+	// - index foreign frames into buckets; here the foreign sort must be respected
+	buckets := make(map[string]int)
+	for i, mf := range foreign.localFrames {
+		buckets[mf.RefValue] = i
+	}
+
+	var k string
+	aux := make([]FrameRow, 0, 100)
+	local.walkRowsLocal(func(i int, r FrameRow) error {
+		v := r[keyColIx]
+		if v == nil {
+			k = ""
+		} else {
+			k = cast.ToString(v.Get())
+		}
+
+		if _, ok := buckets[k]; ok {
+			aux = append(aux, r)
+		}
+
+		return nil
+	})
+
+	modified = len(local.localFrames[0].Rows) != len(aux)
+	local.localFrames[0].Rows = aux
+
+	return
+
+}

store/rdbms/compose_record_datasource.go

@@ -169,6 +169,11 @@ func (r *recordDatasource) Group(d report.GroupDefinition, name string) (bool, e
 
 	// secondly handle column definitions
 	for _, k := range d.Columns {
+		// - make sure the column results with an aggregated value
+		if !r.isAggregated(k.Def.ASTNode) {
+			return false, fmt.Errorf("group column %s does not aggregate data", k.Name)
+		}
+
 		// - validate columns & functions
 		err = k.Def.Traverse(func(n *qlng.ASTNode) (bool, *qlng.ASTNode, error) {
 			if n.Symbol != "" {
@@ -648,3 +653,15 @@ func (r *recordDatasource) validateSort(def *report.FrameDefinition) (canPage bo
 	}()
 	return unique != "", nil
 }
+
+func (r *recordDatasource) isAggregated(n *qlng.ASTNode) bool {
+	switch strings.ToLower(n.Ref) {
+	case "count",
+		"sum",
+		"max",
+		"min",
+		"avg":
+		return true
+	}
+	return false
+}

system/service/report.go

@@ -368,7 +368,7 @@ func (svc *report) RunFresh(ctx context.Context, src types.ReportDataSourceSet,
 
 			// 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 != "") {
+			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
 			}