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

Merged in feature/listtypes (pull request #23)

Browse Source 
Feature/listtypes

* started

* cleanuplist

* someunittestfixes

* removemostflakiness
This commit is contained in:
Michael McGuinness
2025-01-15 12:19:49 +00:00
parent 7da57058d4
commit 5ca36b0502
38 changed files with 532 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/database/repository/collector.sql.go
+23 -14
View File
@@ -16,13 +16,16 @@ SELECT id, jobId, minCleanVersion, minTextVersion FROM collectors WHERE jobId =
`
type GetCollectorFromJobIDRow struct {
ID pgtype.UUID
Jobid pgtype.UUID
Mincleanversion int32
Mintextversion int32
ID pgtype.UUID `db:"id"`
Jobid pgtype.UUID `db:"jobid"`
Mincleanversion int32 `db:"mincleanversion"`
Mintextversion int32 `db:"mintextversion"`
}
func (q *Queries) GetCollectorFromJobID(ctx context.Context, jobid pgtype.UUID) (GetCollectorFromJobIDRow, error) {
// GetCollectorFromJobID
//
// SELECT id, jobId, minCleanVersion, minTextVersion FROM collectors WHERE jobId = $1 LIMIT 1
func (q *Queries) GetCollectorFromJobID(ctx context.Context, jobid pgtype.UUID) (*GetCollectorFromJobIDRow, error) {
row := q.db.QueryRow(ctx, getCollectorFromJobID, jobid)
var i GetCollectorFromJobIDRow
err := row.Scan(
@@ -31,7 +34,7 @@ func (q *Queries) GetCollectorFromJobID(ctx context.Context, jobid pgtype.UUID)
&i.Mincleanversion,
&i.Mintextversion,
)
return i, err
return &i, err
}
const getCollectorQueries = `-- name: GetCollectorQueries :many
@@ -42,20 +45,26 @@ SELECT collectorId, queryId, type, queryVersion, ARRAY_AGG(requiredQueryId) AS r
`
type GetCollectorQueriesRow struct {
Collectorid pgtype.UUID
Queryid pgtype.UUID
Type NullQuerytype
Queryversion pgtype.Int4
Requiredids []pgtype.UUID
Collectorid pgtype.UUID `db:"collectorid"`
Queryid pgtype.UUID `db:"queryid"`
Type NullQuerytype `db:"type"`
Queryversion *int32 `db:"queryversion"`
Requiredids []pgtype.UUID `db:"requiredids"`
}
func (q *Queries) GetCollectorQueries(ctx context.Context, collectorid pgtype.UUID) ([]GetCollectorQueriesRow, error) {
// GetCollectorQueries
//
// SELECT collectorId, queryId, type, queryVersion, ARRAY_AGG(requiredQueryId) AS requiredIds
// FROM collectorQueryDependencyTree
// WHERE collectorId = $1
// GROUP BY queryId, collectorId, type, queryVersion
func (q *Queries) GetCollectorQueries(ctx context.Context, collectorid pgtype.UUID) ([]*GetCollectorQueriesRow, error) {
rows, err := q.db.Query(ctx, getCollectorQueries, collectorid)
if err != nil {
return nil, err
}
defer rows.Close()
var items []GetCollectorQueriesRow
items := []*GetCollectorQueriesRow{}
for rows.Next() {
var i GetCollectorQueriesRow
if err := rows.Scan(
@@ -67,7 +76,7 @@ func (q *Queries) GetCollectorQueries(ctx context.Context, collectorid pgtype.UU
); err != nil {
return nil, err
}
items = append(items, i)
items = append(items, &i)
}
if err := rows.Err(); err != nil {
return nil, err
internal/database/repository/models.go
+64 -55
View File
@@ -53,90 +53,99 @@ func (ns NullQuerytype) Value() (driver.Value, error) {
return string(ns.Querytype), nil
}
func (e Querytype) Valid() bool {
switch e {
case QuerytypeContextFull,
QuerytypeJsonExtractor:
return true
}
return false
}
type Activecollectorquery struct {
Collectorid pgtype.UUID
Activeversion int32
Queryid pgtype.UUID
Collectorid pgtype.UUID `db:"collectorid"`
Activeversion int32 `db:"activeversion"`
Queryid pgtype.UUID `db:"queryid"`
}
type Activequeryrequirement struct {
ID pgtype.UUID
Type Querytype
Activeversion int32
Requiredqueryid pgtype.UUID
ID pgtype.UUID `db:"id"`
Type Querytype `db:"type"`
Activeversion int32 `db:"activeversion"`
Requiredqueryid pgtype.UUID `db:"requiredqueryid"`
}
type Collector struct {
ID pgtype.UUID
Jobid pgtype.UUID
Mincleanversion int32
Mintextversion int32
Latestversion int32
Activeversion int32
ID pgtype.UUID `db:"id"`
Jobid pgtype.UUID `db:"jobid"`
Mincleanversion int32 `db:"mincleanversion"`
Mintextversion int32 `db:"mintextversion"`
Latestversion int32 `db:"latestversion"`
Activeversion int32 `db:"activeversion"`
}
type Collectorquery struct {
ID pgtype.UUID
Collectorid pgtype.UUID
Name string
Queryid pgtype.UUID
Addedversion int32
Removedversion pgtype.Int4
ID pgtype.UUID `db:"id"`
Collectorid pgtype.UUID `db:"collectorid"`
Name string `db:"name"`
Queryid pgtype.UUID `db:"queryid"`
Addedversion int32 `db:"addedversion"`
Removedversion *int32 `db:"removedversion"`
}
type Collectorquerydependencytree struct {
Collectorid pgtype.UUID
ID pgtype.UUID
Type Querytype
Requiredqueryid pgtype.UUID
Queryversion int32
Collectorid pgtype.UUID `db:"collectorid"`
ID pgtype.UUID `db:"id"`
Type Querytype `db:"type"`
Requiredqueryid pgtype.UUID `db:"requiredqueryid"`
Queryversion int32 `db:"queryversion"`
}
type Fullactivequery struct {
ID pgtype.UUID
Type Querytype
Activeversion int32
Latestversion int32
Config []byte
Requiredids []pgtype.UUID
ID pgtype.UUID `db:"id"`
Type Querytype `db:"type"`
Activeversion int32 `db:"activeversion"`
Latestversion int32 `db:"latestversion"`
Config []byte `db:"config"`
Requiredids []pgtype.UUID `db:"requiredids"`
}
type Query struct {
ID pgtype.UUID
Latestversion int32
Activeversion int32
Type Querytype
ID pgtype.UUID `db:"id"`
Latestversion int32 `db:"latestversion"`
Activeversion int32 `db:"activeversion"`
Type Querytype `db:"type"`
}
type Queryconfig struct {
ID pgtype.UUID
Queryid pgtype.UUID
Config []byte
Addedversion int32
Removedversion pgtype.Int4
ID pgtype.UUID `db:"id"`
Queryid pgtype.UUID `db:"queryid"`
Config []byte `db:"config"`
Addedversion int32 `db:"addedversion"`
Removedversion *int32 `db:"removedversion"`
}
type Querydeprecation struct {
ID pgtype.UUID
Queryid pgtype.UUID
Time pgtype.Timestamp
Removedat pgtype.Timestamp
ID pgtype.UUID `db:"id"`
Queryid pgtype.UUID `db:"queryid"`
Time pgtype.Timestamp `db:"time"`
Removedat pgtype.Timestamp `db:"removedat"`
}
type Requiredquery struct {
ID pgtype.UUID
Queryid pgtype.UUID
Requiredqueryid pgtype.UUID
Addedversion int32
Removedversion pgtype.Int4
ID pgtype.UUID `db:"id"`
Queryid pgtype.UUID `db:"queryid"`
Requiredqueryid pgtype.UUID `db:"requiredqueryid"`
Addedversion int32 `db:"addedversion"`
Removedversion *int32 `db:"removedversion"`
}
type Result struct {
ID pgtype.UUID
Queryid pgtype.UUID
Documentid pgtype.UUID
Value string
Cleanversion int32
Textversion int32
Queryversion int32
ID pgtype.UUID `db:"id"`
Queryid pgtype.UUID `db:"queryid"`
Documentid pgtype.UUID `db:"documentid"`
Value string `db:"value"`
Cleanversion int32 `db:"cleanversion"`
Textversion int32 `db:"textversion"`
Queryversion int32 `db:"queryversion"`
}
internal/database/repository/query.sql.go
+45 -19
View File
@@ -15,6 +15,9 @@ const createQuery = `-- name: CreateQuery :one
INSERT INTO queries (type) VALUES ($1) RETURNING id
`
// CreateQuery
//
// INSERT INTO queries (type) VALUES ($1) RETURNING id
func (q *Queries) CreateQuery(ctx context.Context, type_ Querytype) (pgtype.UUID, error) {
row := q.db.QueryRow(ctx, createQuery, type_)
var id pgtype.UUID
@@ -27,12 +30,15 @@ INSERT INTO queryConfigs (queryId, config, addedVersion) VALUES ($1, $2, $3)
`
type CreateQueryConfigParams struct {
Queryid pgtype.UUID
Config []byte
Addedversion int32
Queryid pgtype.UUID `db:"queryid"`
Config []byte `db:"config"`
Addedversion int32 `db:"addedversion"`
}
func (q *Queries) CreateQueryConfig(ctx context.Context, arg CreateQueryConfigParams) error {
// CreateQueryConfig
//
// INSERT INTO queryConfigs (queryId, config, addedVersion) VALUES ($1, $2, $3)
func (q *Queries) CreateQueryConfig(ctx context.Context, arg *CreateQueryConfigParams) error {
_, err := q.db.Exec(ctx, createQueryConfig, arg.Queryid, arg.Config, arg.Addedversion)
return err
}
@@ -42,12 +48,15 @@ INSERT INTO requiredQueries (queryId, requiredQueryId, addedVersion) VALUES ($1,
`
type CreateRequiredQueryParams struct {
Queryid pgtype.UUID
Requiredqueryid pgtype.UUID
Addedversion int32
Queryid pgtype.UUID `db:"queryid"`
Requiredqueryid pgtype.UUID `db:"requiredqueryid"`
Addedversion int32 `db:"addedversion"`
}
func (q *Queries) CreateRequiredQuery(ctx context.Context, arg CreateRequiredQueryParams) error {
// CreateRequiredQuery
//
// INSERT INTO requiredQueries (queryId, requiredQueryId, addedVersion) VALUES ($1, $2, $3)
func (q *Queries) CreateRequiredQuery(ctx context.Context, arg *CreateRequiredQueryParams) error {
_, err := q.db.Exec(ctx, createRequiredQuery, arg.Queryid, arg.Requiredqueryid, arg.Addedversion)
return err
}
@@ -56,6 +65,9 @@ const deprecateQuery = `-- name: DeprecateQuery :exec
INSERT INTO queryDeprecations (queryId) VALUES ($1)
`
// DeprecateQuery
//
// INSERT INTO queryDeprecations (queryId) VALUES ($1)
func (q *Queries) DeprecateQuery(ctx context.Context, queryid pgtype.UUID) error {
_, err := q.db.Exec(ctx, deprecateQuery, queryid)
return err
@@ -65,7 +77,10 @@ const getQuery = `-- name: GetQuery :one
SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries WHERE id = $1
`
func (q *Queries) GetQuery(ctx context.Context, id pgtype.UUID) (Fullactivequery, error) {
// GetQuery
//
// SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries WHERE id = $1
func (q *Queries) GetQuery(ctx context.Context, id pgtype.UUID) (*Fullactivequery, error) {
row := q.db.QueryRow(ctx, getQuery, id)
var i Fullactivequery
err := row.Scan(
@@ -76,7 +91,7 @@ func (q *Queries) GetQuery(ctx context.Context, id pgtype.UUID) (Fullactivequery
&i.Config,
&i.Requiredids,
)
return i, err
return &i, err
}
const getQueryConfig = `-- name: GetQueryConfig :one
@@ -84,20 +99,23 @@ SELECT id, config FROM queryConfigs where queryId = $1 and addedVersion >= $2 an
`
type GetQueryConfigParams struct {
Queryid pgtype.UUID
Addedversion int32
Queryid pgtype.UUID `db:"queryid"`
Addedversion int32 `db:"addedversion"`
}
type GetQueryConfigRow struct {
ID pgtype.UUID
Config []byte
ID pgtype.UUID `db:"id"`
Config []byte `db:"config"`
}
func (q *Queries) GetQueryConfig(ctx context.Context, arg GetQueryConfigParams) (GetQueryConfigRow, error) {
// GetQueryConfig
//
// SELECT id, config FROM queryConfigs where queryId = $1 and addedVersion >= $2 and COALESCE(removedVersion, $2 - 1) < $2
func (q *Queries) GetQueryConfig(ctx context.Context, arg *GetQueryConfigParams) (*GetQueryConfigRow, error) {
row := q.db.QueryRow(ctx, getQueryConfig, arg.Queryid, arg.Addedversion)
var i GetQueryConfigRow
err := row.Scan(&i.ID, &i.Config)
return i, err
return &i, err
}
const isQueryDeprecated = `-- name: IsQueryDeprecated :one
@@ -106,6 +124,11 @@ SELECT EXISTS (
)
`
// IsQueryDeprecated
//
// SELECT EXISTS (
// SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is null LIMIT 1
// )
func (q *Queries) IsQueryDeprecated(ctx context.Context, queryid pgtype.UUID) (bool, error) {
row := q.db.QueryRow(ctx, isQueryDeprecated, queryid)
var exists bool
@@ -117,13 +140,16 @@ const listQueries = `-- name: ListQueries :many
SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries
`
func (q *Queries) ListQueries(ctx context.Context) ([]Fullactivequery, error) {
// ListQueries
//
// SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries
func (q *Queries) ListQueries(ctx context.Context) ([]*Fullactivequery, error) {
rows, err := q.db.Query(ctx, listQueries)
if err != nil {
return nil, err
}
defer rows.Close()
var items []Fullactivequery
items := []*Fullactivequery{}
for rows.Next() {
var i Fullactivequery
if err := rows.Scan(
@@ -136,7 +162,7 @@ func (q *Queries) ListQueries(ctx context.Context) ([]Fullactivequery, error) {
); err != nil {
return nil, err
}
items = append(items, i)
items = append(items, &i)
}
if err := rows.Err(); err != nil {
return nil, err
internal/database/repository/query_test.go
+44 -24
View File
@@ -2,7 +2,6 @@ package repository_test
import (
"context"
"log"
"os"
"path"
"queryorchestration/internal/database"
@@ -35,14 +34,14 @@ func TestQueries(t *testing.T) {
jsonConfig := []byte("{\"path\": \"example_path\"}")
err = queries.CreateRequiredQuery(ctx, repository.CreateRequiredQueryParams{
err = queries.CreateRequiredQuery(ctx, &repository.CreateRequiredQueryParams{
Queryid: jsonQueryID,
Requiredqueryid: contextQueryID,
Addedversion: 1,
})
assert.Nil(t, err)
err = queries.CreateQueryConfig(ctx, repository.CreateQueryConfigParams{
err = queries.CreateQueryConfig(ctx, &repository.CreateQueryConfigParams{
Queryid: jsonQueryID,
Config: jsonConfig,
Addedversion: 1,
@@ -58,34 +57,15 @@ func TestQueries(t *testing.T) {
Latestversion: 1,
Config: jsonConfig,
Requiredids: []pgtype.UUID{contextQueryID},
}, jsonQuery)
}, *jsonQuery)
jsonQueryConfig, err := queries.GetQueryConfig(ctx, repository.GetQueryConfigParams{
jsonQueryConfig, err := queries.GetQueryConfig(ctx, &repository.GetQueryConfigParams{
Queryid: jsonQueryID,
Addedversion: jsonQuery.Activeversion,
})
assert.Nil(t, err)
assert.Equal(t, jsonConfig, jsonQueryConfig.Config)
qs, err := queries.ListQueries(ctx)
assert.Nil(t, err)
assert.Len(t, qs, 2)
log.Print(qs)
assert.ElementsMatch(t, []repository.Fullactivequery{
{ID: jsonQueryID,
Type: repository.QuerytypeJsonExtractor,
Activeversion: 1,
Latestversion: 1,
Config: jsonConfig,
Requiredids: []pgtype.UUID{contextQueryID}},
{ID: contextQueryID,
Type: repository.QuerytypeContextFull,
Activeversion: 1,
Latestversion: 1,
Config: nil,
Requiredids: []pgtype.UUID{{}}},
}, qs)
isDeprecated, err := queries.IsQueryDeprecated(ctx, jsonQueryID)
assert.Nil(t, err)
assert.False(t, isDeprecated)
@@ -97,3 +77,43 @@ func TestQueries(t *testing.T) {
assert.Nil(t, err)
assert.True(t, isDeprecated)
}
func TestQueriesList(t *testing.T) {
ctx := context.Background()
db, cleanup := test.CreateDB(t, ctx, &test.CreateDatabaseConfig{
Migrations: &database.MigrationConfig{
BasePath: "../../..",
}})
defer cleanup()
queries := repository.New(db.Pool)
contextQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull))
assert.Nil(t, err)
jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor))
assert.Nil(t, err)
qs, err := queries.ListQueries(ctx)
assert.Nil(t, err)
assert.Len(t, qs, 2)
assert.ElementsMatch(t, []*repository.Fullactivequery{
{
ID: jsonQueryID,
Type: repository.QuerytypeJsonExtractor,
Activeversion: 1,
Latestversion: 1,
Config: nil,
Requiredids: []pgtype.UUID{{}},
},
{
ID: contextQueryID,
Type: repository.QuerytypeContextFull,
Activeversion: 1,
Latestversion: 1,
Config: nil,
Requiredids: []pgtype.UUID{{}},
},
}, qs)
}
internal/database/repository/result.sql.go
+31 -22
View File
@@ -16,24 +16,27 @@ SELECT id, queryId, value FROM results where id = ANY($1)
`
type ListResultValuesByIDRow struct {
ID pgtype.UUID
Queryid pgtype.UUID
Value string
ID pgtype.UUID `db:"id"`
Queryid pgtype.UUID `db:"queryid"`
Value string `db:"value"`
}
func (q *Queries) ListResultValuesByID(ctx context.Context, id []pgtype.UUID) ([]ListResultValuesByIDRow, error) {
// ListResultValuesByID
//
// SELECT id, queryId, value FROM results where id = ANY($1)
func (q *Queries) ListResultValuesByID(ctx context.Context, id []pgtype.UUID) ([]*ListResultValuesByIDRow, error) {
rows, err := q.db.Query(ctx, listResultValuesByID, id)
if err != nil {
return nil, err
}
defer rows.Close()
var items []ListResultValuesByIDRow
items := []*ListResultValuesByIDRow{}
for rows.Next() {
var i ListResultValuesByIDRow
if err := rows.Scan(&i.ID, &i.Queryid, &i.Value); err != nil {
return nil, err
}
items = append(items, i)
items = append(items, &i)
}
if err := rows.Err(); err != nil {
return nil, err
@@ -46,30 +49,33 @@ SELECT id, queryId, queryVersion FROM results where documentId = $1 and cleanVer
`
type ListResultsByDocumentIDParams struct {
Documentid pgtype.UUID
Cleanversion int32
Textversion int32
Documentid pgtype.UUID `db:"documentid"`
Cleanversion int32 `db:"cleanversion"`
Textversion int32 `db:"textversion"`
}
type ListResultsByDocumentIDRow struct {
ID pgtype.UUID
Queryid pgtype.UUID
Queryversion int32
ID pgtype.UUID `db:"id"`
Queryid pgtype.UUID `db:"queryid"`
Queryversion int32 `db:"queryversion"`
}
func (q *Queries) ListResultsByDocumentID(ctx context.Context, arg ListResultsByDocumentIDParams) ([]ListResultsByDocumentIDRow, error) {
// ListResultsByDocumentID
//
// SELECT id, queryId, queryVersion FROM results where documentId = $1 and cleanVersion >= $2 and textVersion >= $3
func (q *Queries) ListResultsByDocumentID(ctx context.Context, arg *ListResultsByDocumentIDParams) ([]*ListResultsByDocumentIDRow, error) {
rows, err := q.db.Query(ctx, listResultsByDocumentID, arg.Documentid, arg.Cleanversion, arg.Textversion)
if err != nil {
return nil, err
}
defer rows.Close()
var items []ListResultsByDocumentIDRow
items := []*ListResultsByDocumentIDRow{}
for rows.Next() {
var i ListResultsByDocumentIDRow
if err := rows.Scan(&i.ID, &i.Queryid, &i.Queryversion); err != nil {
return nil, err
}
items = append(items, i)
items = append(items, &i)
}
if err := rows.Err(); err != nil {
return nil, err
@@ -82,15 +88,18 @@ INSERT INTO results (queryId, documentId, value, cleanVersion, textVersion, quer
`
type SetResultParams struct {
Queryid pgtype.UUID
Documentid pgtype.UUID
Value string
Cleanversion int32
Textversion int32
Queryversion int32
Queryid pgtype.UUID `db:"queryid"`
Documentid pgtype.UUID `db:"documentid"`
Value string `db:"value"`
Cleanversion int32 `db:"cleanversion"`
Textversion int32 `db:"textversion"`
Queryversion int32 `db:"queryversion"`
}
func (q *Queries) SetResult(ctx context.Context, arg SetResultParams) (pgtype.UUID, error) {
// SetResult
//
// INSERT INTO results (queryId, documentId, value, cleanVersion, textVersion, queryVersion) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id
func (q *Queries) SetResult(ctx context.Context, arg *SetResultParams) (pgtype.UUID, error) {
row := q.db.QueryRow(ctx, setResult,
arg.Queryid,
arg.Documentid,
internal/database/repository/result_test.go
+4 -4
View File
@@ -37,7 +37,7 @@ func TestResults(t *testing.T) {
textVersion := int32(1)
jsonResultValue := "example_value"
jsonResultID, err := queries.SetResult(ctx, repository.SetResultParams{
jsonResultID, err := queries.SetResult(ctx, &repository.SetResultParams{
Queryid: jsonQueryID,
Documentid: documentID,
Value: jsonResultValue,
@@ -48,14 +48,14 @@ func TestResults(t *testing.T) {
assert.Nil(t, err)
assert.True(t, jsonResultID.Valid)
resultsByDoc, err := queries.ListResultsByDocumentID(ctx, repository.ListResultsByDocumentIDParams{
resultsByDoc, err := queries.ListResultsByDocumentID(ctx, &repository.ListResultsByDocumentIDParams{
Documentid: documentID,
Cleanversion: cleanVersion,
Textversion: textVersion,
})
assert.Nil(t, err)
assert.Len(t, resultsByDoc, 1)
assert.ElementsMatch(t, []repository.ListResultsByDocumentIDRow{
assert.ElementsMatch(t, []*repository.ListResultsByDocumentIDRow{
{
ID: jsonResultID,
Queryid: jsonQueryID,
@@ -66,7 +66,7 @@ func TestResults(t *testing.T) {
results, err := queries.ListResultValuesByID(ctx, []pgtype.UUID{jsonResultID})
assert.Nil(t, err)
assert.Len(t, results, 1)
assert.ElementsMatch(t, []repository.ListResultValuesByIDRow{
assert.ElementsMatch(t, []*repository.ListResultValuesByIDRow{
{
ID: jsonResultID,
Queryid: jsonQueryID,