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

Merged in feature/testwithlogs (pull request #65)

Browse Source 
Query Version Sync Runner

* testing

* queryversiosyncworking

* update

* tests

* fixtests
This commit is contained in:
Michael McGuinness
2025-02-14 10:56:24 +00:00
parent 477518e5eb
commit 0df3d16976
91 changed files with 1737 additions and 624 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/database/repository/query.sql.go
+80 -15
View File
@@ -125,21 +125,40 @@ func (q *Queries) GetQueryConfig(ctx context.Context, arg *GetQueryConfigParams)
}
const getQueryWithVersion = `-- name: GetQueryWithVersion :one
SELECT DISTINCT q.id, q.type, q.activeVersion, q.latestVersion, coalesce(c.config, null) as config, ARRAY_AGG(DISTINCT r.requiredQueryId)::uuid[] as requiredIds
FROM queries AS q
WITH query as (
SELECT id, type, activeVersion, latestVersion FROM queries WHERE id = $1
),
config as (
SELECT c.queryId, c.config
FROM query AS q
LEFT JOIN queryConfigs AS c ON q.id = c.queryId
and $2 >= c.addedVersion
and $2 < COALESCE(c.removedVersion, $2 + 1)
and (c.removedVersion is null or $2 < c.removedVersion)
),
requiredIds as (
SELECT r.queryId,
ARRAY_AGG(DISTINCT r.requiredQueryId)
FILTER (WHERE r.requiredQueryId != '00000000-0000-0000-0000-000000000000')::uuid[]
as requiredIds
FROM query AS q
LEFT JOIN requiredQueries AS r ON q.id = r.queryId
and $2 >= r.addedVersion
and $2 < COALESCE(r.removedVersion, $2 + 1)
WHERE q.id = $1
GROUP BY q.id, q.type, q.activeversion, q.latestversion, c.config
and (r.removedVersion is null or $2 < r.removedVersion)
GROUP BY r.queryId
)
SELECT DISTINCT q.id, q.type, q.activeVersion, q.latestVersion, c.config,
coalesce(
r.requiredIds,
array[]::uuid[]
)::uuid[] as requiredIds
FROM query AS q
LEFT JOIN config AS c ON q.id = c.queryId
LEFT JOIN requiredIds AS r ON q.id = r.queryId
`
type GetQueryWithVersionParams struct {
ID pgtype.UUID `db:"id"`
Addedversion int32 `db:"addedversion"`
ID pgtype.UUID `db:"id"`
Version *int32 `db:"version"`
}
type GetQueryWithVersionRow struct {
@@ -153,18 +172,37 @@ type GetQueryWithVersionRow struct {
// GetQueryWithVersion
//
// SELECT DISTINCT q.id, q.type, q.activeVersion, q.latestVersion, coalesce(c.config, null) as config, ARRAY_AGG(DISTINCT r.requiredQueryId)::uuid[] as requiredIds
// FROM queries AS q
// WITH query as (
// SELECT id, type, activeVersion, latestVersion FROM queries WHERE id = $1
// ),
// config as (
// SELECT c.queryId, c.config
// FROM query AS q
// LEFT JOIN queryConfigs AS c ON q.id = c.queryId
// and $2 >= c.addedVersion
// and $2 < COALESCE(c.removedVersion, $2 + 1)
// and (c.removedVersion is null or $2 < c.removedVersion)
// ),
// requiredIds as (
// SELECT r.queryId,
// ARRAY_AGG(DISTINCT r.requiredQueryId)
// FILTER (WHERE r.requiredQueryId != '00000000-0000-0000-0000-000000000000')::uuid[]
// as requiredIds
// FROM query AS q
// LEFT JOIN requiredQueries AS r ON q.id = r.queryId
// and $2 >= r.addedVersion
// and $2 < COALESCE(r.removedVersion, $2 + 1)
// WHERE q.id = $1
// GROUP BY q.id, q.type, q.activeversion, q.latestversion, c.config
// and (r.removedVersion is null or $2 < r.removedVersion)
// GROUP BY r.queryId
// )
// SELECT DISTINCT q.id, q.type, q.activeVersion, q.latestVersion, c.config,
// coalesce(
// r.requiredIds,
// array[]::uuid[]
// )::uuid[] as requiredIds
// FROM query AS q
// LEFT JOIN config AS c ON q.id = c.queryId
// LEFT JOIN requiredIds AS r ON q.id = r.queryId
func (q *Queries) GetQueryWithVersion(ctx context.Context, arg *GetQueryWithVersionParams) (*GetQueryWithVersionRow, error) {
row := q.db.QueryRow(ctx, getQueryWithVersion, arg.ID, arg.Addedversion)
row := q.db.QueryRow(ctx, getQueryWithVersion, arg.ID, arg.Version)
var i GetQueryWithVersionRow
err := row.Scan(
&i.ID,
@@ -312,6 +350,33 @@ func (q *Queries) ListQueryDirectDependentsByDocumentID(ctx context.Context, arg
return items, nil
}
const listQueryJobIDs = `-- name: ListQueryJobIDs :many
SELECT jobId FROM collectorQueryDependencyTree WHERE queryId = $1
`
// ListQueryJobIDs
//
// SELECT jobId FROM collectorQueryDependencyTree WHERE queryId = $1
func (q *Queries) ListQueryJobIDs(ctx context.Context, queryid pgtype.UUID) ([]pgtype.UUID, error) {
rows, err := q.db.Query(ctx, listQueryJobIDs, queryid)
if err != nil {
return nil, err
}
defer rows.Close()
items := []pgtype.UUID{}
for rows.Next() {
var jobid pgtype.UUID
if err := rows.Scan(&jobid); err != nil {
return nil, err
}
items = append(items, jobid)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const removeQueryConfig = `-- name: RemoveQueryConfig :exec
UPDATE queryConfigs SET removedVersion = $1 WHERE queryId = $2 and removedVersion is null
`
internal/database/repository/query_test.go
+111 -2
View File
@@ -67,6 +67,17 @@ func TestQueries(t *testing.T) {
})
assert.NoError(t, err)
jsonQuery, err = queries.GetQuery(ctx, jsonQueryID)
assert.NoError(t, err)
assert.EqualExportedValues(t, &repository.Fullactivequery{
ID: jsonQueryID,
Type: repository.QuerytypeJsonExtractor,
Activeversion: 1,
Latestversion: 2,
Config: nil,
Requiredids: []pgtype.UUID{contextQueryID},
}, jsonQuery)
removeV := int32(2)
err = queries.RemoveRequiredQuery(ctx, &repository.RemoveRequiredQueryParams{
Queryid: jsonQueryID,
@@ -131,9 +142,10 @@ func TestQueries(t *testing.T) {
Requiredids: []pgtype.UUID{},
}, jsonQuery)
v := int32(1)
versionedQuery, err := queries.GetQueryWithVersion(ctx, &repository.GetQueryWithVersionParams{
ID: jsonQueryID,
Addedversion: 1,
ID: jsonQueryID,
Version: &v,
})
assert.NoError(t, err)
assert.EqualExportedValues(t, &repository.GetQueryWithVersionRow{
@@ -420,3 +432,100 @@ func TestQueriesList(t *testing.T) {
},
}, qs)
}
func TestListQueryJobs(t *testing.T) {
if testing.Short() {
t.Skip("Skipping long test in short mode")
}
ctx := context.Background()
cfg := &serviceconfig.BaseConfig{}
test.SetCfgProvider(t, cfg)
cfg.SetBasePath(path.Join(os.Getenv("PWD"), "../../.."))
_, cleanup := test.CreateDB(t, ctx, &test.CreateDatabaseConfig{
Cfg: cfg,
RunMigrations: true,
})
defer cleanup()
queries := cfg.GetDBQueries()
contextID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull))
assert.NoError(t, err)
jobs, err := queries.ListQueryJobIDs(ctx, contextID)
assert.NoError(t, err)
assert.ElementsMatch(t, []pgtype.UUID{}, jobs)
clientOneID, err := queries.CreateClient(ctx, "example_client")
assert.NoError(t, err)
jobOneID, err := queries.CreateJob(ctx, clientOneID)
assert.NoError(t, err)
collOneID, err := queries.CreateCollector(ctx, jobOneID)
assert.NoError(t, err)
err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
Collectorid: collOneID,
Queryid: contextID,
Addedversion: 1,
Name: "example_key",
})
assert.NoError(t, err)
jobs, err = queries.ListQueryJobIDs(ctx, contextID)
assert.NoError(t, err)
assert.ElementsMatch(t, []pgtype.UUID{jobOneID}, jobs)
jobTwoID, err := queries.CreateJob(ctx, clientOneID)
assert.NoError(t, err)
collTwoID, err := queries.CreateCollector(ctx, jobTwoID)
assert.NoError(t, err)
err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
Collectorid: collTwoID,
Queryid: contextID,
Addedversion: 1,
Name: "example_key",
})
assert.NoError(t, err)
jobs, err = queries.ListQueryJobIDs(ctx, contextID)
assert.NoError(t, err)
assert.ElementsMatch(t, []pgtype.UUID{jobOneID, jobTwoID}, jobs)
clientTwoID, err := queries.CreateClient(ctx, "example_client_two")
assert.NoError(t, err)
jobThreeID, err := queries.CreateJob(ctx, clientTwoID)
assert.NoError(t, err)
collThreeID, err := queries.CreateCollector(ctx, jobThreeID)
assert.NoError(t, err)
err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
Collectorid: collThreeID,
Queryid: contextID,
Addedversion: 1,
Name: "example_key",
})
assert.NoError(t, err)
jobs, err = queries.ListQueryJobIDs(ctx, contextID)
assert.NoError(t, err)
assert.ElementsMatch(t, []pgtype.UUID{jobOneID, jobTwoID, jobThreeID}, jobs)
jsonID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor))
assert.NoError(t, err)
err = queries.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
Queryid: jsonID,
Requiredqueryid: contextID,
Addedversion: 1,
})
assert.NoError(t, err)
err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
Collectorid: collOneID,
Queryid: jsonID,
Addedversion: 1,
Name: "example_key",
})
assert.NoError(t, err)
jobs, err = queries.ListQueryJobIDs(ctx, jsonID)
assert.NoError(t, err)
assert.ElementsMatch(t, []pgtype.UUID{jobOneID}, jobs)
}
internal/database/repository/result.sql.go
+16 -10
View File
@@ -53,17 +53,21 @@ WITH reqQueries as (
and $3 >= rq.addedVersion
and (rq.removedVersion is null or $3 < rq.removedVersion)
),
docs as (
SELECT id, jobId
FROM documents
WHERE id = $1
),
codeVersions as (
SELECT
d.id as documentId,
coalesce(ccv.minCleanVersion, 1) as minCleanVersion,
coalesce(ccv.minTextVersion, 1) as minTextVersion
FROM documents as d
FROM docs as d
LEFT JOIN collectors as c on c.jobId = d.jobId
LEFT JOIN collectorCodeVersions as ccv on c.id = ccv.collectorId
and c.activeVersion >= ccv.addedVersion
and c.activeVersion < COALESCE(ccv.removedVersion, c.activeVersion)
WHERE d.id = $1
LIMIT 1
),
latestVersions AS (
@@ -80,13 +84,12 @@ latestVersions AS (
FROM reqQueries as rq
LEFT JOIN results as r ON rq.queryId = r.queryId
and r.documentId = $1
JOIN codeVersions as ccv on ccv.documentId = r.documentId
WHERE r.documentId = $1
and r.queryVersion = rq.activeVersion
JOIN codeVersions as ccv on ccv.documentId = r.documentId
and r.cleanVersion >= ccv.minCleanVersion
and r.textVersion >= ccv.minTextVersion
)
SELECT lv.queryId, lv.type, r.value
SELECT DISTINCT lv.queryId, lv.type, r.value
FROM latestVersions as lv
JOIN results as r ON r.queryId = lv.queryId
and r.documentId = $1
@@ -118,17 +121,21 @@ type ListQueryRequirementValuesRow struct {
// and $3 >= rq.addedVersion
// and (rq.removedVersion is null or $3 < rq.removedVersion)
// ),
// docs as (
// SELECT id, jobId
// FROM documents
// WHERE id = $1
// ),
// codeVersions as (
// SELECT
// d.id as documentId,
// coalesce(ccv.minCleanVersion, 1) as minCleanVersion,
// coalesce(ccv.minTextVersion, 1) as minTextVersion
// FROM documents as d
// FROM docs as d
// LEFT JOIN collectors as c on c.jobId = d.jobId
// LEFT JOIN collectorCodeVersions as ccv on c.id = ccv.collectorId
// and c.activeVersion >= ccv.addedVersion
// and c.activeVersion < COALESCE(ccv.removedVersion, c.activeVersion)
// WHERE d.id = $1
// LIMIT 1
// ),
// latestVersions AS (
@@ -145,13 +152,12 @@ type ListQueryRequirementValuesRow struct {
// FROM reqQueries as rq
// LEFT JOIN results as r ON rq.queryId = r.queryId
// and r.documentId = $1
// JOIN codeVersions as ccv on ccv.documentId = r.documentId
// WHERE r.documentId = $1
// and r.queryVersion = rq.activeVersion
// JOIN codeVersions as ccv on ccv.documentId = r.documentId
// and r.cleanVersion >= ccv.minCleanVersion
// and r.textVersion >= ccv.minTextVersion
// )
// SELECT lv.queryId, lv.type, r.value
// SELECT DISTINCT lv.queryId, lv.type, r.value
// FROM latestVersions as lv
// JOIN results as r ON r.queryId = lv.queryId
// and r.documentId = $1
internal/database/repository/result_test.go
-1
View File
@@ -398,5 +398,4 @@ func TestUnsyncedNoDepsQueries(t *testing.T) {
assert.NoError(t, err)
assert.Len(t, qs, 1)
assert.ElementsMatch(t, []pgtype.UUID{contextQueryID}, qs)
}