jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

creatorupdatordeprecate

Browse Source
This commit is contained in:
Michael McGuinness
2025-01-03 13:41:07 +00:00
parent 0ac156f0e1
commit 788b21594c
36 changed files with 545 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/queryQueue/create.go
+18 -12
View File
@@ -3,7 +3,7 @@ package queryQueue
import (
"context"
"queryorchestration/internal/database"
"queryorchestration/internal/query"
queryprocessor "queryorchestration/internal/queryProcessor"
)
func (q *Queue) getUnsyncedQueries() {
@@ -36,9 +36,9 @@ func (c *Queue) getCollectorQueries(ctx context.Context) error {
return err
}
cleanQueries := make([]query.QueryRow, len(queries))
cleanQueries := make([]queryprocessor.Query, len(queries))
for index, dbQuery := range queries {
cleanQuery, err := query.ParseDBQueryRow(&dbQuery)
cleanQuery, err := queryprocessor.ParseDBQuery(&dbQuery)
if err != nil {
return err
}
@@ -51,33 +51,39 @@ func (c *Queue) getCollectorQueries(ctx context.Context) error {
return nil
}
func (q *Queue) Add(qu *query.QueryRow) {
dependentQueries := []query.QueryRow{}
func (q *Queue) Add(qu *queryprocessor.Query) {
dependentQueries := []queryprocessor.Query{}
requiredIndex := -1
if q.unsyncedQueue == nil {
q.unsyncedQueue = &[]query.QueryRow{}
q.unsyncedQueue = &[]queryprocessor.Query{}
} else {
for index, entry := range *q.unsyncedQueue {
if entry.ID == qu.ID {
return
}
if entry.ID == qu.RequiredQueryID {
requiredIndex = index
for _, id := range qu.RequiredQueryIDs {
if entry.ID == id {
requiredIndex = index
break
}
}
}
}
for _, entry := range *q.collectorQueries {
if entry.RequiredQueryID == qu.ID {
dependentQueries = append(dependentQueries, entry)
for _, id := range entry.RequiredQueryIDs {
if qu.ID == id {
dependentQueries = append(dependentQueries, entry)
break
}
}
}
if requiredIndex != -1 {
*q.unsyncedQueue = append((*q.unsyncedQueue)[:requiredIndex+1], append([]query.QueryRow{*qu}, (*q.unsyncedQueue)[requiredIndex+1:]...)...)
*q.unsyncedQueue = append((*q.unsyncedQueue)[:requiredIndex+1], append([]queryprocessor.Query{*qu}, (*q.unsyncedQueue)[requiredIndex+1:]...)...)
} else {
*q.unsyncedQueue = append([]query.QueryRow{*qu}, *q.unsyncedQueue...)
*q.unsyncedQueue = append([]queryprocessor.Query{*qu}, *q.unsyncedQueue...)
}
for _, entry := range dependentQueries {
internal/queryQueue/execute.go
+7 -15
View File
@@ -7,10 +7,9 @@ import (
"queryorchestration/internal/database"
"queryorchestration/internal/database/repository"
jsonextractor "queryorchestration/internal/jsonExtractor"
"queryorchestration/internal/query"
queryprocessor "queryorchestration/internal/queryProcessor"
"queryorchestration/internal/result"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
)
@@ -31,16 +30,9 @@ func (q *Queue) Execute(ctx context.Context) error {
return nil
}
func (q *Queue) executeQuery(ctx context.Context, qu query.QueryRow) error {
requiredQueryIDs := []uuid.UUID{}
for _, entry := range *q.collectorQueries {
if entry.ID == qu.ID && entry.RequiredQueryID != uuid.Nil {
requiredQueryIDs = append(requiredQueryIDs, entry.RequiredQueryID)
}
}
resultIDs := make([]pgtype.UUID, len(requiredQueryIDs))
for index, id := range requiredQueryIDs {
func (q *Queue) executeQuery(ctx context.Context, qu queryprocessor.Query) error {
resultIDs := make([]pgtype.UUID, len(qu.RequiredQueryIDs))
for index, id := range qu.RequiredQueryIDs {
var queryVersion int32
for _, entry := range *q.collectorQueries {
if entry.ID == id {
@@ -81,7 +73,7 @@ func (q *Queue) executeQuery(ctx context.Context, qu query.QueryRow) error {
}
func (q *Queue) getResultValue(res *repository.ListResultValuesByIDRow) (result.Value, error) {
var queryType query.Type
var queryType queryprocessor.Type
for _, qu := range *q.collectorQueries {
if qu.ID == database.MustToUUID(res.Queryid) {
queryType = qu.Type
@@ -89,9 +81,9 @@ func (q *Queue) getResultValue(res *repository.ListResultValuesByIDRow) (result.
}
switch queryType {
case query.TypeJsonExtractor:
case queryprocessor.TypeJsonExtractor:
return jsonextractor.NewResult(res.Value), nil
case query.TypeContextFull:
case queryprocessor.TypeContextFull:
return contextfull.NewResult(res.Value), nil
default:
return nil, fmt.Errorf("attempting to process invalid query type")
internal/queryQueue/result.go
+7 -7
View File
@@ -5,11 +5,11 @@ import (
"fmt"
contextfull "queryorchestration/internal/contextFull"
jsonextractor "queryorchestration/internal/jsonExtractor"
"queryorchestration/internal/query"
queryprocessor "queryorchestration/internal/queryProcessor"
"queryorchestration/internal/result"
)
func (q *Queue) setResult(ctx context.Context, qu query.QueryRow, resultValues *[]result.Value) error {
func (q *Queue) setResult(ctx context.Context, qu queryprocessor.Query, resultValues *[]result.Value) error {
processor, err := q.getProcessor(qu.Type)
if err != nil {
return err
@@ -41,12 +41,12 @@ func (q *Queue) setResult(ctx context.Context, qu query.QueryRow, resultValues *
return nil
}
func (q *Queue) getProcessor(queryType query.Type) (query.Processor, error) {
func (q *Queue) getProcessor(queryType queryprocessor.Type) (queryprocessor.Processor, error) {
switch queryType {
case query.TypeJsonExtractor:
return jsonextractor.New(q.db), nil
case query.TypeContextFull:
return contextfull.New(), nil
case queryprocessor.TypeJsonExtractor:
return jsonextractor.NewExtractor(q.db), nil
case queryprocessor.TypeContextFull:
return contextfull.NewExtractor(), nil
default:
return nil, fmt.Errorf("attempting to process invalid query type")
}
internal/queryQueue/service.go
+4 -4
View File
@@ -4,15 +4,15 @@ import (
"context"
"queryorchestration/internal/collector"
"queryorchestration/internal/database/repository"
"queryorchestration/internal/query"
queryprocessor "queryorchestration/internal/queryProcessor"
"queryorchestration/internal/result"
"github.com/google/uuid"
)
type Queue struct {
unsyncedQueue *[]query.QueryRow
collectorQueries *[]query.QueryRow
unsyncedQueue *[]queryprocessor.Query
collectorQueries *[]queryprocessor.Query
results *[]result.Result
collector *collector.Collector
db *repository.Queries
@@ -41,6 +41,6 @@ func New(ctx context.Context, db *repository.Queries, coll *collector.Collector,
return &queue, nil
}
func (q *Queue) GetQueue() []query.QueryRow {
func (q *Queue) GetQueue() []queryprocessor.Query {
return *q.unsyncedQueue
}