Merge pull request #26 from NeedleInAJayStack/feature/parallelize-hisRead

fix: parallelizes HisRead with Filter

Author: NeedleInAJayStack

pkg/plugin/datasource.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/NeedleInAJayStack/haystack"
@@ -164,14 +166,21 @@ func (datasource *Datasource) query(ctx context.Context, pCtx backend.PluginCont
 		return response
 
 	case "hisReadFilter":
-		points, readErr := datasource.read(model.HisReadFilter, variables)
+		pointsGrid, readErr := datasource.read(model.HisReadFilter+" and hisStart", variables)
 		if readErr != nil {
 			log.DefaultLogger.Error(readErr.Error())
 			return backend.ErrDataResponse(backend.StatusBadRequest, fmt.Sprintf("HisReadFilter failure: %v", readErr.Error()))
 		}
+		points := pointsGrid.Rows()
+		recLimit := 300
+		if len(points) > recLimit {
+			errMsg := fmt.Sprintf("Query exceeded record limit of %d: %d records", recLimit, len(points))
+			log.DefaultLogger.Error(errMsg)
+			return backend.ErrDataResponse(backend.StatusBadRequest, errMsg)
+		}
 
-		grids := []haystack.Grid{}
-		for _, point := range points.Rows() {
+		// Function to read a single point and send it to a channel.
+		readPoint := func(point haystack.Row, hisReadChannel chan haystack.Grid, wg *sync.WaitGroup) {
 			id := point.Get("id")
 			var ref haystack.Ref
 			switch id.(type) {
@@ -180,15 +189,38 @@ func (datasource *Datasource) query(ctx context.Context, pCtx backend.PluginCont
 			default:
 				errMsg := fmt.Sprintf("id is not a ref: %v", id)
 				log.DefaultLogger.Error(errMsg)
-				return backend.ErrDataResponse(backend.StatusBadRequest, errMsg)
+				hisReadChannel <- haystack.EmptyGrid()
 			}
 			hisRead, err := datasource.hisRead(ref, query.TimeRange)
 			if err != nil {
 				log.DefaultLogger.Error(err.Error())
-				return backend.ErrDataResponse(backend.StatusBadRequest, fmt.Sprintf("HisReadFilter failure: %v", err.Error()))
+				hisReadChannel <- haystack.EmptyGrid()
 			}
-			grids = append(grids, hisRead)
+			hisReadChannel <- hisRead
+			wg.Done()
 		}
+
+		// Start a goroutine to collect all the grids into a slice.
+		hisReadChannel := make(chan haystack.Grid)
+		combinedChannel := make(chan []haystack.Grid)
+		go func() {
+			grids := []haystack.Grid{}
+			for grid := range hisReadChannel {
+				grids = append(grids, grid)
+			}
+			combinedChannel <- grids
+		}()
+
+		// Read all the points in parallel using goroutines.
+		var wg sync.WaitGroup
+		wg.Add(len(points))
+		for _, point := range points {
+			go readPoint(point, hisReadChannel, &wg)
+		}
+		wg.Wait()
+		close(hisReadChannel)
+
+		grids := <-combinedChannel
 		response := responseFromGrids(grids)
 		// Make the display name on the "val" fields the names of the points.
 		for _, frame := range response.Frames {
@@ -213,19 +245,25 @@ func (datasource *Datasource) query(ctx context.Context, pCtx backend.PluginCont
 	}
 }
 
+// Creates a response from the input grids. The frames in the result are sorted by display name.
 func responseFromGrids(grids []haystack.Grid) backend.DataResponse {
-	var response backend.DataResponse
+	frames := data.Frames{}
 	for _, grid := range grids {
 		frame, frameErr := dataFrameFromGrid(grid)
 		if frameErr != nil {
 			log.DefaultLogger.Error(frameErr.Error())
 			return backend.ErrDataResponse(backend.StatusBadRequest, fmt.Sprintf("Frame conversion failure: %v", frameErr.Error()))
 		}
-
-		// add the frames to the response.
-		response.Frames = append(response.Frames, frame)
-		response.Status = backend.StatusOK
+		frames = append(frames, frame)
 	}
+
+	sort.Slice(frames, func(i, j int) bool {
+		return frames[i].Name < frames[j].Name
+	})
+
+	var response backend.DataResponse
+	response.Frames = frames
+	response.Status = backend.StatusOK
 	return response
 }