Merged in feature/remove-query (pull request #201)
remove query from codebase part 1 * remove query * fix localstack run
This commit is contained in:
Jay Brown
@@ -0,0 +1,287 @@
|
||||
-- Down migration to restore query functionality
|
||||
-- This recreates all query-related tables, views, and functions
|
||||
|
||||
-- Drop the simplified fullActiveCollectors view
|
||||
DROP VIEW IF EXISTS fullActiveCollectors;
|
||||
|
||||
-- Recreate query type enum
|
||||
CREATE TYPE queryType AS ENUM ('context_full', 'json_extractor');
|
||||
|
||||
-- Recreate queries table
|
||||
CREATE TABLE queries (
|
||||
queryId uuid primary key DEFAULT uuid_generate_v7(),
|
||||
queryType queryType not null
|
||||
);
|
||||
|
||||
-- Recreate queryVersions table
|
||||
CREATE TABLE queryVersions (
|
||||
queryId uuid not null,
|
||||
versionId int not null,
|
||||
addedAt timestamp not null default current_timestamp,
|
||||
primary key (versionId, queryId),
|
||||
foreign key (queryId) references queries(queryId)
|
||||
);
|
||||
|
||||
-- Recreate version number function and trigger
|
||||
CREATE OR REPLACE FUNCTION setQueryVersionNumber()
|
||||
RETURNS TRIGGER AS $$
|
||||
BEGIN
|
||||
SELECT COALESCE(MAX(versionId), 0) + 1
|
||||
INTO NEW.versionId
|
||||
FROM queryVersions
|
||||
WHERE queryId = NEW.queryId;
|
||||
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
|
||||
CREATE TRIGGER setQueryVersionNumberTrigger
|
||||
BEFORE INSERT ON queryVersions
|
||||
FOR EACH ROW
|
||||
EXECUTE FUNCTION setQueryVersionNumber();
|
||||
|
||||
-- Recreate queryActiveVersions table
|
||||
CREATE TABLE queryActiveVersions (
|
||||
activeVersionEntryId uuid primary key DEFAULT uuid_generate_v7(),
|
||||
queryId uuid not null,
|
||||
versionId int not null,
|
||||
foreign key (queryId, versionId) references queryVersions(queryId, versionId)
|
||||
);
|
||||
|
||||
-- Recreate requiredQueries table
|
||||
CREATE TABLE requiredQueries (
|
||||
requiredQueryEntryId uuid primary key DEFAULT uuid_generate_v7(),
|
||||
queryId uuid not null,
|
||||
requiredQueryId uuid not null,
|
||||
addedVersion int not null,
|
||||
removedVersion int,
|
||||
foreign key (queryId) references queries(queryId),
|
||||
foreign key (requiredQueryId) references queries(queryId),
|
||||
foreign key (queryId, addedVersion) references queryVersions(queryId, versionId),
|
||||
foreign key (queryId, removedVersion) references queryVersions(queryId, versionId),
|
||||
unique (queryId, requiredQueryId, removedVersion)
|
||||
);
|
||||
|
||||
-- Recreate queryConfigs table
|
||||
CREATE TABLE queryConfigs (
|
||||
configId uuid primary key DEFAULT uuid_generate_v7(),
|
||||
queryId uuid not null,
|
||||
config jsonb not null,
|
||||
addedVersion int not null,
|
||||
removedVersion int,
|
||||
foreign key (queryId) references queries(queryId),
|
||||
foreign key (queryId, addedVersion) references queryVersions(queryId, versionId),
|
||||
foreign key (queryId, removedVersion) references queryVersions(queryId, versionId),
|
||||
unique (queryId, removedVersion)
|
||||
);
|
||||
|
||||
-- Recreate removeQueryConfig function and trigger
|
||||
CREATE OR REPLACE FUNCTION removeQueryConfig()
|
||||
RETURNS TRIGGER AS $$
|
||||
BEGIN
|
||||
UPDATE queryConfigs
|
||||
SET removedVersion = NEW.addedVersion
|
||||
WHERE queryId = NEW.queryId and removedVersion is null;
|
||||
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
|
||||
CREATE TRIGGER removeQueryConfigTrigger
|
||||
BEFORE INSERT ON queryConfigs
|
||||
FOR EACH ROW
|
||||
EXECUTE FUNCTION removeQueryConfig();
|
||||
|
||||
-- Recreate collectorQueries table
|
||||
CREATE TABLE collectorQueries (
|
||||
id uuid primary key DEFAULT uuid_generate_v7(),
|
||||
clientId varchar(255) not null,
|
||||
name varchar(255) not null,
|
||||
queryId uuid not null,
|
||||
addedVersion int not null,
|
||||
removedVersion int,
|
||||
foreign key (queryId) references queries(queryId),
|
||||
foreign key (clientId) references clients(clientId),
|
||||
foreign key (clientId, addedVersion) references collectorVersions(clientId, id),
|
||||
foreign key (clientId, removedVersion) references collectorVersions(clientId, id),
|
||||
unique (clientId, name, removedVersion)
|
||||
);
|
||||
|
||||
-- Recreate results table
|
||||
CREATE TABLE results (
|
||||
id uuid primary key DEFAULT uuid_generate_v7(),
|
||||
textEntryId uuid not null,
|
||||
queryId uuid not null,
|
||||
value TEXT not null,
|
||||
queryVersion int not null,
|
||||
foreign key (queryId) references queries(queryId),
|
||||
foreign key (textEntryId) references documentTextExtractions(id),
|
||||
foreign key (queryId, queryVersion) references queryVersions(queryId, versionId)
|
||||
);
|
||||
|
||||
-- Recreate resultDependencies table
|
||||
CREATE TABLE resultDependencies (
|
||||
resultId uuid not null,
|
||||
requiredResultId uuid not null,
|
||||
foreign key (resultId) references results(id),
|
||||
foreign key (requiredResultId) references results(id),
|
||||
CONSTRAINT result_not_self_dependent CHECK (resultId != requiredResultId)
|
||||
);
|
||||
|
||||
-- Recreate query views
|
||||
CREATE VIEW queryCurrentActiveVersions as
|
||||
SELECT DISTINCT
|
||||
q.queryId,
|
||||
coalesce(
|
||||
(FIRST_VALUE(av.versionId) OVER (PARTITION BY q.queryId ORDER BY av.activeVersionEntryId DESC)),
|
||||
0
|
||||
)::int as activeVersion
|
||||
FROM queries AS q
|
||||
LEFT JOIN queryActiveVersions as av on av.queryId = q.queryId;
|
||||
|
||||
CREATE VIEW queryLatestVersions as
|
||||
SELECT
|
||||
q.queryId,
|
||||
coalesce(max(v.versionId), 0)::int as latestVersion
|
||||
FROM queries AS q
|
||||
LEFT JOIN queryVersions as v on v.queryId = q.queryId
|
||||
GROUP BY q.queryId;
|
||||
|
||||
CREATE VIEW queryCurrentConfigs as
|
||||
SELECT av.queryId, c.config
|
||||
FROM queryCurrentActiveVersions as av
|
||||
LEFT JOIN queryConfigs AS c ON av.queryId = c.queryId
|
||||
and isInVersion(av.activeVersion, c.addedVersion, c.removedVersion);
|
||||
|
||||
CREATE VIEW queryCurrentRequiredIds as
|
||||
SELECT DISTINCT av.queryId, r.requiredQueryId
|
||||
FROM queryCurrentActiveVersions as av
|
||||
LEFT JOIN requiredQueries AS r ON av.queryId = r.queryId
|
||||
and isInVersion(av.activeVersion, r.addedVersion, r.removedVersion);
|
||||
|
||||
CREATE VIEW queryCurrentRequiredIdsAGG as
|
||||
SELECT queryId,
|
||||
coalesce(
|
||||
ARRAY_AGG(DISTINCT requiredQueryId)
|
||||
FILTER (WHERE requiredQueryId != '00000000-0000-0000-0000-000000000000')::uuid[],
|
||||
array[]::uuid[]
|
||||
)::uuid[] as requiredIds
|
||||
FROM queryCurrentRequiredIds
|
||||
GROUP BY queryId;
|
||||
|
||||
CREATE VIEW fullActiveQueries AS
|
||||
SELECT DISTINCT q.queryId, q.queryType, av.activeVersion, lv.latestVersion, c.config, r.requiredIds
|
||||
FROM queries AS q
|
||||
JOIN queryCurrentActiveVersions as av on q.queryId = av.queryId
|
||||
JOIN queryLatestVersions as lv on lv.queryId = q.queryId
|
||||
JOIN queryCurrentConfigs AS c ON q.queryId = c.queryId
|
||||
JOIN queryCurrentRequiredIdsAGG AS r ON q.queryId = r.queryId;
|
||||
|
||||
CREATE VIEW queryActiveDependencies AS
|
||||
WITH RECURSIVE queryActiveDependencies(queryId, requiredQueryId, path, cycle) AS (
|
||||
SELECT
|
||||
queryId,
|
||||
requiredQueryId,
|
||||
ARRAY[queryId, requiredQueryId]::uuid[] AS path,
|
||||
false AS cycle
|
||||
FROM queryCurrentRequiredIds
|
||||
|
||||
UNION ALL
|
||||
|
||||
SELECT
|
||||
q.queryId,
|
||||
qd.requiredQueryId,
|
||||
path || qd.requiredQueryId,
|
||||
qd.requiredQueryId = ANY(path) AS cycle
|
||||
FROM queryCurrentRequiredIds as q
|
||||
JOIN queryActiveDependencies as qd ON q.queryId = qd.requiredQueryId
|
||||
WHERE NOT qd.cycle
|
||||
)
|
||||
SELECT DISTINCT queryId as id, requiredQueryId
|
||||
FROM queryActiveDependencies
|
||||
WHERE NOT cycle;
|
||||
|
||||
-- Recreate collector views that depend on queries
|
||||
CREATE VIEW currentCollectorQueries as
|
||||
SELECT DISTINCT
|
||||
av.clientId,
|
||||
q.name,
|
||||
q.queryId
|
||||
FROM collectorCurrentActiveVersions as av
|
||||
LEFT JOIN collectorQueries AS q ON av.clientId = q.clientId
|
||||
and isInVersion(av.activeVersion, q.addedVersion, q.removedVersion);
|
||||
|
||||
CREATE VIEW currentCollectorQueriesJSONAGG as
|
||||
SELECT DISTINCT
|
||||
clientId,
|
||||
jsonb_object_agg(name, queryId) FILTER (WHERE name is not null) AS fields
|
||||
FROM currentCollectorQueries
|
||||
GROUP BY clientId;
|
||||
|
||||
CREATE VIEW fullActiveCollectors AS
|
||||
SELECT DISTINCT av.clientId,
|
||||
ccv.minCleanVersion, ctv.minTextVersion,
|
||||
av.activeVersion, lv.latestVersion,
|
||||
q.fields
|
||||
FROM collectorCurrentActiveVersions as av
|
||||
JOIN collectorLatestVersions as lv on lv.clientId = av.clientId
|
||||
JOIN currentCollectorMinCleanVersions AS ccv ON av.clientId = ccv.clientId
|
||||
JOIN currentCollectorMinTextVersions AS ctv ON av.clientId = ctv.clientId
|
||||
JOIN currentCollectorQueriesJSONAGG AS q ON av.clientId = q.clientId;
|
||||
|
||||
CREATE VIEW collectorQueryDependencyTree AS
|
||||
WITH RECURSIVE collectorQueryDependencyTree AS (
|
||||
SELECT cq.clientId, cq.queryId, ri.requiredIds
|
||||
FROM currentCollectorQueries as cq
|
||||
JOIN queryCurrentRequiredIdsAGG as ri on cq.queryId = ri.queryId
|
||||
|
||||
UNION ALL
|
||||
|
||||
SELECT acq.clientId, q.queryId, q.requiredIds
|
||||
FROM queryCurrentRequiredIdsAGG as q
|
||||
JOIN collectorQueryDependencyTree as acq on q.queryId = ANY(acq.requiredIds)
|
||||
)
|
||||
SELECT DISTINCT ct.clientId, ct.queryId, q.queryType, av.activeVersion as queryVersion, ct.requiredIds
|
||||
FROM collectorQueryDependencyTree as ct
|
||||
JOIN queryCurrentActiveVersions as av on ct.queryId = av.queryId
|
||||
JOIN queries as q on q.queryId = ct.queryId;
|
||||
|
||||
CREATE OR REPLACE FUNCTION collectorQueryDependencyTreeByClient(
|
||||
_clientId varchar(255)
|
||||
)
|
||||
RETURNS TABLE (
|
||||
clientId varchar(255),
|
||||
queryId uuid,
|
||||
queryType queryType,
|
||||
queryVersion int,
|
||||
requiredIds uuid[]
|
||||
) AS $$
|
||||
BEGIN
|
||||
RETURN QUERY
|
||||
WITH clientQueries as (
|
||||
SELECT q.clientId, q.queryId
|
||||
FROM currentCollectorQueries as q
|
||||
WHERE q.clientId = _clientId
|
||||
),
|
||||
dependencyTree as (
|
||||
WITH RECURSIVE collectorQueryDependencyTree AS (
|
||||
SELECT cq.clientId, cq.queryId, ri.requiredIds
|
||||
FROM clientQueries as cq
|
||||
JOIN queryCurrentRequiredIdsAGG as ri on cq.queryId = ri.queryId
|
||||
|
||||
UNION ALL
|
||||
|
||||
SELECT acq.clientId, q.queryId, q.requiredIds
|
||||
FROM queryCurrentRequiredIdsAGG as q
|
||||
JOIN collectorQueryDependencyTree as acq on q.queryId = ANY(acq.requiredIds)
|
||||
)
|
||||
SELECT DISTINCT ct.clientId, ct.queryId, q.queryType, av.activeVersion as queryVersion, ct.requiredIds
|
||||
FROM collectorQueryDependencyTree as ct
|
||||
JOIN queryCurrentActiveVersions as av on ct.queryId = av.queryId
|
||||
JOIN queries as q on q.queryId = ct.queryId
|
||||
)
|
||||
SELECT t.clientId, t.queryId, t.queryType, t.queryVersion, t.requiredIds
|
||||
FROM dependencyTree as t
|
||||
WHERE t.clientID is not null;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
@@ -0,0 +1,53 @@
|
||||
-- Migration to remove query functionality from the database
|
||||
-- This drops all query-related tables, views, and functions
|
||||
|
||||
-- First drop the function that depends on query views
|
||||
DROP FUNCTION IF EXISTS collectorQueryDependencyTreeByClient;
|
||||
|
||||
-- Drop the collector views that depend on query tables (in order of dependencies)
|
||||
DROP VIEW IF EXISTS collectorQueryDependencyTree;
|
||||
DROP VIEW IF EXISTS fullActiveCollectors;
|
||||
DROP VIEW IF EXISTS currentCollectorQueriesJSONAGG;
|
||||
DROP VIEW IF EXISTS currentCollectorQueries;
|
||||
|
||||
-- Drop query views (in order of dependencies)
|
||||
DROP VIEW IF EXISTS fullActiveQueries;
|
||||
DROP VIEW IF EXISTS queryActiveDependencies;
|
||||
DROP VIEW IF EXISTS queryCurrentRequiredIdsAGG;
|
||||
DROP VIEW IF EXISTS queryCurrentRequiredIds;
|
||||
DROP VIEW IF EXISTS queryCurrentConfigs;
|
||||
DROP VIEW IF EXISTS queryLatestVersions;
|
||||
DROP VIEW IF EXISTS queryCurrentActiveVersions;
|
||||
|
||||
-- Drop result tables
|
||||
DROP TABLE IF EXISTS resultDependencies;
|
||||
DROP TABLE IF EXISTS results;
|
||||
|
||||
-- Drop collectorQueries table (links collectors to queries)
|
||||
DROP TABLE IF EXISTS collectorQueries;
|
||||
|
||||
-- Drop query configuration and dependency tables
|
||||
DROP TABLE IF EXISTS queryConfigs;
|
||||
DROP TABLE IF EXISTS requiredQueries;
|
||||
DROP TABLE IF EXISTS queryActiveVersions;
|
||||
DROP TABLE IF EXISTS queryVersions;
|
||||
DROP TABLE IF EXISTS queries;
|
||||
|
||||
-- Drop triggers and functions
|
||||
DROP TRIGGER IF EXISTS setQueryVersionNumberTrigger ON queryVersions;
|
||||
DROP FUNCTION IF EXISTS setQueryVersionNumber;
|
||||
DROP TRIGGER IF EXISTS removeQueryConfigTrigger ON queryConfigs;
|
||||
DROP FUNCTION IF EXISTS removeQueryConfig;
|
||||
|
||||
-- Drop the query type enum
|
||||
DROP TYPE IF EXISTS queryType;
|
||||
|
||||
-- Recreate the fullActiveCollectors view without the fields column
|
||||
CREATE VIEW fullActiveCollectors AS
|
||||
SELECT DISTINCT av.clientId,
|
||||
ccv.minCleanVersion, ctv.minTextVersion,
|
||||
av.activeVersion, lv.latestVersion
|
||||
FROM collectorCurrentActiveVersions as av
|
||||
JOIN collectorLatestVersions as lv on lv.clientId = av.clientId
|
||||
JOIN currentCollectorMinCleanVersions AS ccv ON av.clientId = ccv.clientId
|
||||
JOIN currentCollectorMinTextVersions AS ctv ON av.clientId = ctv.clientId;
|
||||
@@ -15,6 +15,9 @@ UPDATE clients SET name = $1 WHERE clientId = $2;
|
||||
INSERT INTO clientCanSync (canSync, clientId) VALUES ($1, $2);
|
||||
|
||||
-- name: IsClientSynced :one
|
||||
-- Query functionality has been removed. A client is considered synced when
|
||||
-- all documents have completed text extraction (or failed with explicit failure).
|
||||
-- See remove_query_plan.md for details.
|
||||
WITH
|
||||
docs AS (
|
||||
-- Get all documents for this client
|
||||
@@ -42,46 +45,6 @@ doc_text_entries AS (
|
||||
FROM
|
||||
doc_clean_entries d
|
||||
LEFT JOIN currentTextEntries cte ON cte.cleanId = d.clean_entry_id
|
||||
),
|
||||
required_results AS (
|
||||
-- All required document-query-version combinations
|
||||
SELECT
|
||||
dte.document_id,
|
||||
cqdt.queryId,
|
||||
cqdt.queryVersion,
|
||||
dte.text_entry_id
|
||||
FROM
|
||||
doc_text_entries dte
|
||||
JOIN collectorQueryDependencyTreeByClient($1) cqdt ON cqdt.clientId = dte.client_id
|
||||
where dte.clean_fail is null and dte.text_entry_id IS NOT NULL
|
||||
),
|
||||
existing_results AS (
|
||||
-- Valid results that exist
|
||||
SELECT
|
||||
rr.document_id AS document_id,
|
||||
r.queryId,
|
||||
r.id AS result_id
|
||||
FROM
|
||||
results r
|
||||
JOIN required_results rr ON
|
||||
r.queryId = rr.queryId AND
|
||||
r.queryVersion = rr.queryVersion AND
|
||||
r.textEntryId = rr.text_entry_id
|
||||
),
|
||||
missing_results AS (
|
||||
-- Find missing results
|
||||
SELECT rr.queryId
|
||||
FROM required_results rr
|
||||
LEFT JOIN existing_results er on er.queryId = rr.queryId
|
||||
WHERE er.result_id is null
|
||||
),
|
||||
dependency_check AS (
|
||||
-- Check for missing dependencies
|
||||
SELECT DISTINCT er.queryId, er.result_id, qcri.queryId, rd.resultId
|
||||
FROM existing_results er
|
||||
JOIN queryCurrentRequiredIds qcri on qcri.requiredQueryId = er.queryId
|
||||
LEFT JOIN resultDependencies rd on rd.requiredResultId = er.result_id
|
||||
WHERE rd.resultId is null
|
||||
)
|
||||
SELECT (
|
||||
-- No documents means client is synced
|
||||
@@ -89,22 +52,11 @@ SELECT (
|
||||
|
||||
OR
|
||||
|
||||
-- Documents with no text entries
|
||||
(
|
||||
NOT EXISTS (
|
||||
SELECT 1 FROM doc_text_entries
|
||||
where clean_entry_id is null or
|
||||
(clean_fail is null and text_entry_id is null)
|
||||
)
|
||||
|
||||
and
|
||||
|
||||
-- Documents with missing results
|
||||
NOT EXISTS (SELECT 1 FROM missing_results)
|
||||
|
||||
and
|
||||
|
||||
-- Documents with missing dependencies
|
||||
NOT EXISTS (SELECT 1 FROM dependency_check)
|
||||
)::bool
|
||||
-- All documents have completed processing (clean entry exists and either
|
||||
-- text entry exists or clean failed)
|
||||
NOT EXISTS (
|
||||
SELECT 1 FROM doc_text_entries
|
||||
where clean_entry_id is null or
|
||||
(clean_fail is null and text_entry_id is null)
|
||||
)
|
||||
)::bool as is_synced;
|
||||
|
||||
@@ -1,7 +1,3 @@
|
||||
-- name: ListCollectorQueries :many
|
||||
SELECT clientId, queryId, queryType, queryVersion, requiredIds
|
||||
FROM collectorQueryDependencyTreeByClient(@clientId);
|
||||
|
||||
-- name: GetCollectorByClientID :one
|
||||
SELECT * FROM fullActiveCollectors WHERE clientId = @clientId LIMIT 1;
|
||||
|
||||
@@ -16,9 +12,3 @@ INSERT INTO collectorMinCleanVersions (clientId, addedVersion, versionId) VALUES
|
||||
|
||||
-- name: SetCollectorTextVersion :exec
|
||||
INSERT INTO collectorMinTextVersions (clientId, addedVersion, versionId) VALUES ($1, $2, $3);
|
||||
|
||||
-- name: AddCollectorQuery :exec
|
||||
INSERT INTO collectorQueries (clientId, name, queryId, addedVersion) VALUES ($1, $2, $3, $4);
|
||||
|
||||
-- name: RemoveCollectorQuery :exec
|
||||
UPDATE collectorQueries SET removedVersion = $1 WHERE queryId = $2 and clientId = $3 and removedVersion is null;
|
||||
|
||||
@@ -2,30 +2,13 @@
|
||||
SELECT id, clientId, hash FROM documents WHERE id = $1;
|
||||
|
||||
-- name: GetDocumentExternal :one
|
||||
WITH
|
||||
docs AS (
|
||||
SELECT id, hash, clientId
|
||||
FROM documents
|
||||
WHERE id = @documentId
|
||||
),
|
||||
namedResults AS (
|
||||
SELECT
|
||||
q.name,
|
||||
dd.id,
|
||||
d.value
|
||||
FROM currentCollectorQueries AS q
|
||||
JOIN docs as dd on dd.clientId = q.clientId
|
||||
LEFT JOIN listValidDocumentResults(@documentId) AS d
|
||||
ON d.queryId = q.queryId and q.name is not null
|
||||
)
|
||||
-- Query functionality has been removed. See remove_query_plan.md for details.
|
||||
SELECT
|
||||
d.id,
|
||||
d.clientId,
|
||||
d.hash,
|
||||
COALESCE(jsonb_object_agg(r.name, r.value) FILTER (WHERE r.name IS NOT NULL), '{}') AS fields
|
||||
FROM docs AS d
|
||||
LEFT JOIN namedResults AS r ON d.id = r.id
|
||||
GROUP BY d.id, d.clientId, d.hash;
|
||||
id,
|
||||
clientId,
|
||||
hash
|
||||
FROM documents
|
||||
WHERE id = @documentId;
|
||||
|
||||
-- name: ListDocumentsByClient :many
|
||||
SELECT id, hash from documents where clientId = @clientId;
|
||||
|
||||
@@ -1,86 +0,0 @@
|
||||
-- name: GetActiveQueryConfig :one
|
||||
SELECT config FROM queryCurrentConfigs where queryId = $1;
|
||||
|
||||
-- name: GetQuery :one
|
||||
SELECT * FROM fullActiveQueries WHERE queryId = $1;
|
||||
|
||||
-- name: GetQueryWithVersion :one
|
||||
WITH query as (
|
||||
SELECT queryId, queryType FROM queries WHERE queryId = @id
|
||||
),
|
||||
config as (
|
||||
SELECT c.queryId, c.config
|
||||
FROM query AS q
|
||||
LEFT JOIN queryConfigs AS c ON q.queryId = c.queryId
|
||||
and isInVersion(@version, c.addedVersion, c.removedVersion)
|
||||
),
|
||||
requiredIds as (
|
||||
SELECT r.queryId,
|
||||
coalesce(
|
||||
ARRAY_AGG(DISTINCT r.requiredQueryId)
|
||||
FILTER (WHERE r.requiredQueryId != '00000000-0000-0000-0000-000000000000')::uuid[],
|
||||
array[]::uuid[]
|
||||
)::uuid[] as requiredIds
|
||||
FROM query AS q
|
||||
LEFT JOIN requiredQueries AS r ON q.queryId = r.queryId
|
||||
and isInVersion(@version, r.addedVersion, r.removedVersion)
|
||||
GROUP BY r.queryId
|
||||
)
|
||||
SELECT DISTINCT q.queryId, q.queryType,
|
||||
av.activeVersion,
|
||||
lv.latestVersion, c.config,
|
||||
r.requiredIds
|
||||
FROM query AS q
|
||||
JOIN queryCurrentActiveVersions as av on q.queryId = av.queryId
|
||||
JOIN queryLatestVersions as lv on lv.queryId = q.queryId
|
||||
LEFT JOIN config AS c ON q.queryId = c.queryId
|
||||
LEFT JOIN requiredIds AS r ON q.queryId = r.queryId;
|
||||
|
||||
-- name: ListQueries :many
|
||||
SELECT * FROM fullActiveQueries;
|
||||
|
||||
-- name: ListQueriesById :many
|
||||
SELECT * FROM fullActiveQueries WHERE queryId = any($1);
|
||||
|
||||
-- name: CreateQuery :one
|
||||
INSERT INTO queries (queryType) VALUES ($1) RETURNING queryId;
|
||||
|
||||
-- name: AddLatestQueryVersion :one
|
||||
INSERT INTO queryVersions (queryId) VALUES ($1) RETURNING versionId;
|
||||
|
||||
-- name: AddActiveQueryVersion :exec
|
||||
INSERT INTO queryActiveVersions (queryId, versionId) VALUES ($1, $2);
|
||||
|
||||
-- name: AddRequiredQuery :exec
|
||||
INSERT INTO requiredQueries (queryId, requiredQueryId, addedVersion) VALUES ($1, $2, $3);
|
||||
|
||||
-- name: RemoveRequiredQuery :exec
|
||||
UPDATE requiredQueries SET removedVersion = $1 WHERE requiredQueryId = $2 and queryId = $3 and removedVersion is null;
|
||||
|
||||
-- name: SetQueryConfig :exec
|
||||
INSERT INTO queryConfigs (queryId, config, addedVersion) VALUES ($1, $2, $3);
|
||||
|
||||
-- name: AllQueriesExist :one
|
||||
SELECT COUNT(*) = COUNT(DISTINCT queryId) AS all_exist
|
||||
FROM unnest($1::uuid[]) AS input_id
|
||||
LEFT JOIN queries ON input_id = queries.queryId;
|
||||
|
||||
-- name: IsQueryInDependencyTree :one
|
||||
SELECT EXISTS (
|
||||
SELECT 1 FROM queryActiveDependencies
|
||||
WHERE id = any(@requiredQueryIds)
|
||||
and requiredQueryId = @queryId
|
||||
or @queryId = any(@requiredQueryIds)
|
||||
);
|
||||
|
||||
-- name: ListQueryDirectDependentsByDocumentID :many
|
||||
WITH doc AS (
|
||||
SELECT clientId FROM documents WHERE id = @documentId
|
||||
)
|
||||
SELECT dt.queryId
|
||||
FROM doc,
|
||||
LATERAL collectorQueryDependencyTreeByClient(doc.clientId) AS dt
|
||||
WHERE @queryId = ANY(dt.requiredIds);
|
||||
|
||||
-- name: ListQueryClientIDs :many
|
||||
SELECT clientId FROM collectorQueryDependencyTree WHERE queryId = $1;
|
||||
@@ -1,92 +0,0 @@
|
||||
-- name: ListQueryRequirementValues :many
|
||||
WITH reqQueries as (
|
||||
SELECT av.queryId, av.activeVersion, q.queryType
|
||||
FROM requiredQueries as rq
|
||||
JOIN queryCurrentActiveVersions as av on av.queryId = rq.requiredQueryId
|
||||
JOIN queries as q on q.queryId = av.queryId
|
||||
WHERE rq.queryId = @queryId
|
||||
and isInVersion(@version, rq.addedVersion, rq.removedVersion)
|
||||
),
|
||||
docs as (
|
||||
SELECT id, clientId
|
||||
FROM documents
|
||||
WHERE id = @documentId
|
||||
),
|
||||
codeVersions as (
|
||||
SELECT
|
||||
d.id as documentId,
|
||||
mcv.minCleanVersion,
|
||||
mtv.minTextVersion
|
||||
FROM docs as d
|
||||
JOIN currentCollectorMinTextVersions as mtv on mtv.clientId = d.clientId
|
||||
JOIN currentCollectorMinCleanVersions as mcv on mcv.clientId = d.clientId
|
||||
),
|
||||
latestVersions AS (
|
||||
SELECT
|
||||
r.id,
|
||||
rq.queryId,
|
||||
rq.queryType,
|
||||
r.queryVersion,
|
||||
r.textEntryId,
|
||||
r.value,
|
||||
ROW_NUMBER() OVER (
|
||||
PARTITION BY r.queryId
|
||||
ORDER BY r.id DESC
|
||||
) as rowNumber
|
||||
FROM reqQueries as rq
|
||||
JOIN currentTextEntries as cte on cte.documentId = @documentId
|
||||
LEFT JOIN results as r ON rq.queryId = r.queryId
|
||||
and r.queryVersion = rq.activeVersion
|
||||
and cte.id = r.textEntryId
|
||||
)
|
||||
SELECT DISTINCT id, queryId, queryType, value
|
||||
FROM latestVersions
|
||||
WHERE rowNumber = 1;
|
||||
|
||||
-- name: AddResult :one
|
||||
INSERT INTO results (queryId, value, textEntryId, queryVersion) VALUES ($1, $2, $3, $4) returning id;
|
||||
|
||||
-- name: AddResultDependency :exec
|
||||
INSERT INTO resultDependencies (resultId, requiredResultId) VALUES ($1, $2);
|
||||
|
||||
-- name: GetResultValueWithVersion :one
|
||||
WITH doc as (
|
||||
SELECT id, clientId
|
||||
FROM documents
|
||||
WHERE id = @documentId
|
||||
)
|
||||
SELECT r.id, r.value
|
||||
FROM doc as d
|
||||
JOIN currentTextEntries as cte on cte.documentId = d.id
|
||||
LEFT JOIN results as r
|
||||
on r.queryId = @queryId
|
||||
and r.queryVersion = @queryVersion
|
||||
and r.textEntryId = cte.id;
|
||||
|
||||
-- name: ListUnsyncedNoDepsQueriesByDocId :many
|
||||
WITH docs as (
|
||||
SELECT id, clientId from documents where id = $1
|
||||
),
|
||||
dependency_tree as (
|
||||
select q.queryId, q.requiredIds, q.clientId, q.queryVersion
|
||||
from docs,
|
||||
LATERAL collectorQueryDependencyTreeByClient(docs.clientId) AS q
|
||||
),
|
||||
text_entries as (
|
||||
select id, documentId from currentTextEntries where documentId = $1
|
||||
),
|
||||
unsyncedQueries AS (
|
||||
SELECT DISTINCT dt.queryId, dt.requiredIds
|
||||
from docs as d
|
||||
JOIN dependency_tree as dt on d.clientId = dt.clientId
|
||||
JOIN text_entries as cte on cte.documentId = d.id
|
||||
LEFT JOIN results as r
|
||||
on r.queryId = dt.queryId
|
||||
and r.queryVersion = dt.queryVersion
|
||||
and cte.id = r.textEntryId
|
||||
where r.value is null
|
||||
)
|
||||
SELECT DISTINCT queryId FROM unsyncedQueries as baseuq
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1 FROM unsyncedQueries as uq WHERE uq.queryId = any(baseuq.requiredIds)
|
||||
);
|
||||
@@ -85,46 +85,6 @@ doc_text_entries AS (
|
||||
FROM
|
||||
doc_clean_entries d
|
||||
LEFT JOIN currentTextEntries cte ON cte.cleanId = d.clean_entry_id
|
||||
),
|
||||
required_results AS (
|
||||
-- All required document-query-version combinations
|
||||
SELECT
|
||||
dte.document_id,
|
||||
cqdt.queryId,
|
||||
cqdt.queryVersion,
|
||||
dte.text_entry_id
|
||||
FROM
|
||||
doc_text_entries dte
|
||||
JOIN collectorQueryDependencyTreeByClient($1) cqdt ON cqdt.clientId = dte.client_id
|
||||
where dte.clean_fail is null and dte.text_entry_id IS NOT NULL
|
||||
),
|
||||
existing_results AS (
|
||||
-- Valid results that exist
|
||||
SELECT
|
||||
rr.document_id AS document_id,
|
||||
r.queryId,
|
||||
r.id AS result_id
|
||||
FROM
|
||||
results r
|
||||
JOIN required_results rr ON
|
||||
r.queryId = rr.queryId AND
|
||||
r.queryVersion = rr.queryVersion AND
|
||||
r.textEntryId = rr.text_entry_id
|
||||
),
|
||||
missing_results AS (
|
||||
-- Find missing results
|
||||
SELECT rr.queryId
|
||||
FROM required_results rr
|
||||
LEFT JOIN existing_results er on er.queryId = rr.queryId
|
||||
WHERE er.result_id is null
|
||||
),
|
||||
dependency_check AS (
|
||||
-- Check for missing dependencies
|
||||
SELECT DISTINCT er.queryId, er.result_id, qcri.queryId, rd.resultId
|
||||
FROM existing_results er
|
||||
JOIN queryCurrentRequiredIds qcri on qcri.requiredQueryId = er.queryId
|
||||
LEFT JOIN resultDependencies rd on rd.requiredResultId = er.result_id
|
||||
WHERE rd.resultId is null
|
||||
)
|
||||
SELECT (
|
||||
-- No documents means client is synced
|
||||
@@ -132,28 +92,19 @@ SELECT (
|
||||
|
||||
OR
|
||||
|
||||
-- Documents with no text entries
|
||||
(
|
||||
NOT EXISTS (
|
||||
SELECT 1 FROM doc_text_entries
|
||||
where clean_entry_id is null or
|
||||
(clean_fail is null and text_entry_id is null)
|
||||
)
|
||||
|
||||
and
|
||||
|
||||
-- Documents with missing results
|
||||
NOT EXISTS (SELECT 1 FROM missing_results)
|
||||
|
||||
and
|
||||
|
||||
-- Documents with missing dependencies
|
||||
NOT EXISTS (SELECT 1 FROM dependency_check)
|
||||
)::bool
|
||||
-- All documents have completed processing (clean entry exists and either
|
||||
-- text entry exists or clean failed)
|
||||
NOT EXISTS (
|
||||
SELECT 1 FROM doc_text_entries
|
||||
where clean_entry_id is null or
|
||||
(clean_fail is null and text_entry_id is null)
|
||||
)
|
||||
)::bool as is_synced
|
||||
`
|
||||
|
||||
// IsClientSynced
|
||||
// Query functionality has been removed. A client is considered synced when
|
||||
// all documents have completed text extraction (or failed with explicit failure).
|
||||
// See remove_query_plan.md for details.
|
||||
//
|
||||
// WITH
|
||||
// docs AS (
|
||||
@@ -182,46 +133,6 @@ SELECT (
|
||||
// FROM
|
||||
// doc_clean_entries d
|
||||
// LEFT JOIN currentTextEntries cte ON cte.cleanId = d.clean_entry_id
|
||||
// ),
|
||||
// required_results AS (
|
||||
// -- All required document-query-version combinations
|
||||
// SELECT
|
||||
// dte.document_id,
|
||||
// cqdt.queryId,
|
||||
// cqdt.queryVersion,
|
||||
// dte.text_entry_id
|
||||
// FROM
|
||||
// doc_text_entries dte
|
||||
// JOIN collectorQueryDependencyTreeByClient($1) cqdt ON cqdt.clientId = dte.client_id
|
||||
// where dte.clean_fail is null and dte.text_entry_id IS NOT NULL
|
||||
// ),
|
||||
// existing_results AS (
|
||||
// -- Valid results that exist
|
||||
// SELECT
|
||||
// rr.document_id AS document_id,
|
||||
// r.queryId,
|
||||
// r.id AS result_id
|
||||
// FROM
|
||||
// results r
|
||||
// JOIN required_results rr ON
|
||||
// r.queryId = rr.queryId AND
|
||||
// r.queryVersion = rr.queryVersion AND
|
||||
// r.textEntryId = rr.text_entry_id
|
||||
// ),
|
||||
// missing_results AS (
|
||||
// -- Find missing results
|
||||
// SELECT rr.queryId
|
||||
// FROM required_results rr
|
||||
// LEFT JOIN existing_results er on er.queryId = rr.queryId
|
||||
// WHERE er.result_id is null
|
||||
// ),
|
||||
// dependency_check AS (
|
||||
// -- Check for missing dependencies
|
||||
// SELECT DISTINCT er.queryId, er.result_id, qcri.queryId, rd.resultId
|
||||
// FROM existing_results er
|
||||
// JOIN queryCurrentRequiredIds qcri on qcri.requiredQueryId = er.queryId
|
||||
// LEFT JOIN resultDependencies rd on rd.requiredResultId = er.result_id
|
||||
// WHERE rd.resultId is null
|
||||
// )
|
||||
// SELECT (
|
||||
// -- No documents means client is synced
|
||||
@@ -229,24 +140,13 @@ SELECT (
|
||||
//
|
||||
// OR
|
||||
//
|
||||
// -- Documents with no text entries
|
||||
// (
|
||||
// NOT EXISTS (
|
||||
// SELECT 1 FROM doc_text_entries
|
||||
// where clean_entry_id is null or
|
||||
// (clean_fail is null and text_entry_id is null)
|
||||
// )
|
||||
//
|
||||
// and
|
||||
//
|
||||
// -- Documents with missing results
|
||||
// NOT EXISTS (SELECT 1 FROM missing_results)
|
||||
//
|
||||
// and
|
||||
//
|
||||
// -- Documents with missing dependencies
|
||||
// NOT EXISTS (SELECT 1 FROM dependency_check)
|
||||
// )::bool
|
||||
// -- All documents have completed processing (clean entry exists and either
|
||||
// -- text entry exists or clean failed)
|
||||
// NOT EXISTS (
|
||||
// SELECT 1 FROM doc_text_entries
|
||||
// where clean_entry_id is null or
|
||||
// (clean_fail is null and text_entry_id is null)
|
||||
// )
|
||||
// )::bool as is_synced
|
||||
func (q *Queries) IsClientSynced(ctx context.Context, dollar_1 *string) (bool, error) {
|
||||
row := q.db.QueryRow(ctx, isClientSynced, dollar_1)
|
||||
|
||||
@@ -7,34 +7,8 @@ package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
const addCollectorQuery = `-- name: AddCollectorQuery :exec
|
||||
INSERT INTO collectorQueries (clientId, name, queryId, addedVersion) VALUES ($1, $2, $3, $4)
|
||||
`
|
||||
|
||||
type AddCollectorQueryParams struct {
|
||||
Clientid string `db:"clientid"`
|
||||
Name string `db:"name"`
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Addedversion int32 `db:"addedversion"`
|
||||
}
|
||||
|
||||
// AddCollectorQuery
|
||||
//
|
||||
// INSERT INTO collectorQueries (clientId, name, queryId, addedVersion) VALUES ($1, $2, $3, $4)
|
||||
func (q *Queries) AddCollectorQuery(ctx context.Context, arg *AddCollectorQueryParams) error {
|
||||
_, err := q.db.Exec(ctx, addCollectorQuery,
|
||||
arg.Clientid,
|
||||
arg.Name,
|
||||
arg.Queryid,
|
||||
arg.Addedversion,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const addLatestCollectorVersion = `-- name: AddLatestCollectorVersion :one
|
||||
INSERT INTO collectorVersions (clientId) VALUES ($1) RETURNING id
|
||||
`
|
||||
@@ -50,12 +24,12 @@ func (q *Queries) AddLatestCollectorVersion(ctx context.Context, clientid string
|
||||
}
|
||||
|
||||
const getCollectorByClientID = `-- name: GetCollectorByClientID :one
|
||||
SELECT clientid, mincleanversion, mintextversion, activeversion, latestversion, fields FROM fullActiveCollectors WHERE clientId = $1 LIMIT 1
|
||||
SELECT clientid, mincleanversion, mintextversion, activeversion, latestversion FROM fullActiveCollectors WHERE clientId = $1 LIMIT 1
|
||||
`
|
||||
|
||||
// GetCollectorByClientID
|
||||
//
|
||||
// SELECT clientid, mincleanversion, mintextversion, activeversion, latestversion, fields FROM fullActiveCollectors WHERE clientId = $1 LIMIT 1
|
||||
// SELECT clientid, mincleanversion, mintextversion, activeversion, latestversion FROM fullActiveCollectors WHERE clientId = $1 LIMIT 1
|
||||
func (q *Queries) GetCollectorByClientID(ctx context.Context, clientid string) (*Fullactivecollector, error) {
|
||||
row := q.db.QueryRow(ctx, getCollectorByClientID, clientid)
|
||||
var i Fullactivecollector
|
||||
@@ -65,72 +39,10 @@ func (q *Queries) GetCollectorByClientID(ctx context.Context, clientid string) (
|
||||
&i.Mintextversion,
|
||||
&i.Activeversion,
|
||||
&i.Latestversion,
|
||||
&i.Fields,
|
||||
)
|
||||
return &i, err
|
||||
}
|
||||
|
||||
const listCollectorQueries = `-- name: ListCollectorQueries :many
|
||||
SELECT clientId, queryId, queryType, queryVersion, requiredIds
|
||||
FROM collectorQueryDependencyTreeByClient($1)
|
||||
`
|
||||
|
||||
type ListCollectorQueriesRow struct {
|
||||
Clientid *string `db:"clientid"`
|
||||
Queryid *uuid.UUID `db:"queryid"`
|
||||
Querytype NullQuerytype `db:"querytype"`
|
||||
Queryversion *int32 `db:"queryversion"`
|
||||
Requiredids []uuid.UUID `db:"requiredids"`
|
||||
}
|
||||
|
||||
// ListCollectorQueries
|
||||
//
|
||||
// SELECT clientId, queryId, queryType, queryVersion, requiredIds
|
||||
// FROM collectorQueryDependencyTreeByClient($1)
|
||||
func (q *Queries) ListCollectorQueries(ctx context.Context, clientid string) ([]*ListCollectorQueriesRow, error) {
|
||||
rows, err := q.db.Query(ctx, listCollectorQueries, clientid)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []*ListCollectorQueriesRow{}
|
||||
for rows.Next() {
|
||||
var i ListCollectorQueriesRow
|
||||
if err := rows.Scan(
|
||||
&i.Clientid,
|
||||
&i.Queryid,
|
||||
&i.Querytype,
|
||||
&i.Queryversion,
|
||||
&i.Requiredids,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, &i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const removeCollectorQuery = `-- name: RemoveCollectorQuery :exec
|
||||
UPDATE collectorQueries SET removedVersion = $1 WHERE queryId = $2 and clientId = $3 and removedVersion is null
|
||||
`
|
||||
|
||||
type RemoveCollectorQueryParams struct {
|
||||
Removedversion *int32 `db:"removedversion"`
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Clientid string `db:"clientid"`
|
||||
}
|
||||
|
||||
// RemoveCollectorQuery
|
||||
//
|
||||
// UPDATE collectorQueries SET removedVersion = $1 WHERE queryId = $2 and clientId = $3 and removedVersion is null
|
||||
func (q *Queries) RemoveCollectorQuery(ctx context.Context, arg *RemoveCollectorQueryParams) error {
|
||||
_, err := q.db.Exec(ctx, removeCollectorQuery, arg.Removedversion, arg.Queryid, arg.Clientid)
|
||||
return err
|
||||
}
|
||||
|
||||
const setActiveCollectorVersion = `-- name: SetActiveCollectorVersion :exec
|
||||
INSERT INTO collectorActiveVersions (clientId, versionId) VALUES ($1, $2)
|
||||
`
|
||||
|
||||
@@ -1,263 +0,0 @@
|
||||
package repository_test
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"queryorchestration/internal/database/repository"
|
||||
"queryorchestration/internal/serviceconfig"
|
||||
"queryorchestration/internal/test"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestCollector(t *testing.T) {
|
||||
t.Parallel()
|
||||
if testing.Short() {
|
||||
t.SkipNow()
|
||||
}
|
||||
ctx := t.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
contextId, err := queries.CreateQuery(ctx, repository.QuerytypeContextFull)
|
||||
require.NoError(t, err)
|
||||
jsonId, err := queries.CreateQuery(ctx, repository.QuerytypeJsonExtractor)
|
||||
require.NoError(t, err)
|
||||
version, err := queries.AddLatestQueryVersion(ctx, jsonId)
|
||||
require.NoError(t, err)
|
||||
err = queries.AddActiveQueryVersion(ctx, &repository.AddActiveQueryVersionParams{
|
||||
Queryid: jsonId,
|
||||
Versionid: version,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, int32(1), version)
|
||||
err = queries.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
|
||||
Queryid: jsonId,
|
||||
Requiredqueryid: contextId,
|
||||
Addedversion: version,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
clientId := "EXAMPLE"
|
||||
err = queries.CreateClient(ctx, &repository.CreateClientParams{
|
||||
Name: "example_client",
|
||||
Clientid: clientId,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
minCleanVersion := int64(2)
|
||||
minTextVersion := int64(4)
|
||||
|
||||
coll, err := queries.GetCollectorByClientID(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivecollector{
|
||||
Clientid: clientId,
|
||||
Mincleanversion: 0,
|
||||
Mintextversion: 0,
|
||||
Activeversion: 0,
|
||||
Latestversion: 0,
|
||||
}, coll)
|
||||
|
||||
version, err = queries.AddLatestCollectorVersion(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, int32(1), version)
|
||||
|
||||
coll, err = queries.GetCollectorByClientID(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivecollector{
|
||||
Clientid: clientId,
|
||||
Mincleanversion: 0,
|
||||
Mintextversion: 0,
|
||||
Activeversion: 0,
|
||||
Latestversion: 1,
|
||||
}, coll)
|
||||
|
||||
err = queries.SetActiveCollectorVersion(ctx, &repository.SetActiveCollectorVersionParams{
|
||||
Versionid: 1,
|
||||
Clientid: clientId,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
coll, err = queries.GetCollectorByClientID(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivecollector{
|
||||
Clientid: clientId,
|
||||
Mincleanversion: 0,
|
||||
Mintextversion: 0,
|
||||
Activeversion: 1,
|
||||
Latestversion: 1,
|
||||
}, coll)
|
||||
|
||||
err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
|
||||
Clientid: clientId,
|
||||
Queryid: jsonId,
|
||||
Addedversion: 1,
|
||||
Name: "example_key",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
err = queries.SetCollectorTextVersion(ctx, &repository.SetCollectorTextVersionParams{
|
||||
Clientid: clientId,
|
||||
Addedversion: 1,
|
||||
Versionid: minTextVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
coll, err = queries.GetCollectorByClientID(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivecollector{
|
||||
Clientid: clientId,
|
||||
Mincleanversion: 0,
|
||||
Mintextversion: minTextVersion,
|
||||
Activeversion: 1,
|
||||
Latestversion: 1,
|
||||
Fields: []byte(fmt.Sprintf("{\"example_key\": \"%s\"}", jsonId.String())),
|
||||
}, coll)
|
||||
|
||||
err = queries.SetCollectorCleanVersion(ctx, &repository.SetCollectorCleanVersionParams{
|
||||
Clientid: clientId,
|
||||
Addedversion: 1,
|
||||
Versionid: minCleanVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
coll, err = queries.GetCollectorByClientID(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivecollector{
|
||||
Clientid: clientId,
|
||||
Mincleanversion: minCleanVersion,
|
||||
Mintextversion: minTextVersion,
|
||||
Activeversion: 1,
|
||||
Latestversion: 1,
|
||||
Fields: []byte(fmt.Sprintf("{\"example_key\": \"%s\"}", jsonId.String())),
|
||||
}, coll)
|
||||
|
||||
qs, err := queries.ListCollectorQueries(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 2)
|
||||
jsonV := int32(1)
|
||||
contextV := int32(0)
|
||||
assert.ElementsMatch(t, []*repository.ListCollectorQueriesRow{
|
||||
{
|
||||
Clientid: &clientId,
|
||||
Queryid: &jsonId,
|
||||
Queryversion: &jsonV,
|
||||
Querytype: repository.NullQuerytype{
|
||||
Querytype: repository.QuerytypeJsonExtractor,
|
||||
Valid: true,
|
||||
},
|
||||
Requiredids: []uuid.UUID{contextId},
|
||||
},
|
||||
{
|
||||
Clientid: &clientId,
|
||||
Queryid: &contextId,
|
||||
Queryversion: &contextV,
|
||||
Querytype: repository.NullQuerytype{
|
||||
Querytype: repository.QuerytypeContextFull,
|
||||
Valid: true,
|
||||
},
|
||||
Requiredids: []uuid.UUID{},
|
||||
},
|
||||
}, qs)
|
||||
|
||||
version, err = queries.AddLatestCollectorVersion(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, int32(2), version)
|
||||
|
||||
err = queries.RemoveCollectorQuery(ctx, &repository.RemoveCollectorQueryParams{
|
||||
Clientid: clientId,
|
||||
Queryid: jsonId,
|
||||
Removedversion: &version,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
err = queries.SetActiveCollectorVersion(ctx, &repository.SetActiveCollectorVersionParams{
|
||||
Versionid: 2,
|
||||
Clientid: clientId,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
err = queries.SetCollectorTextVersion(ctx, &repository.SetCollectorTextVersionParams{
|
||||
Clientid: clientId,
|
||||
Versionid: minTextVersion + 1,
|
||||
Addedversion: version,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
coll, err = queries.GetCollectorByClientID(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivecollector{
|
||||
Clientid: clientId,
|
||||
Mincleanversion: minCleanVersion,
|
||||
Mintextversion: minTextVersion + 1,
|
||||
Activeversion: 2,
|
||||
Latestversion: 2,
|
||||
Fields: nil,
|
||||
}, coll)
|
||||
|
||||
err = queries.SetCollectorCleanVersion(ctx, &repository.SetCollectorCleanVersionParams{
|
||||
Clientid: clientId,
|
||||
Versionid: minCleanVersion + 1,
|
||||
Addedversion: version,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
coll, err = queries.GetCollectorByClientID(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivecollector{
|
||||
Clientid: clientId,
|
||||
Mincleanversion: minCleanVersion + 1,
|
||||
Mintextversion: minTextVersion + 1,
|
||||
Activeversion: 2,
|
||||
Latestversion: 2,
|
||||
Fields: nil,
|
||||
}, coll)
|
||||
}
|
||||
|
||||
func BenchmarkListCollectorQueries(b *testing.B) {
|
||||
ctx := b.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(b, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
clientId := "EXAMPLE"
|
||||
err := queries.CreateClient(ctx, &repository.CreateClientParams{
|
||||
Name: "example_client",
|
||||
Clientid: clientId,
|
||||
})
|
||||
require.NoError(b, err)
|
||||
|
||||
b.ResetTimer()
|
||||
|
||||
for b.Loop() {
|
||||
_, _ = queries.ListCollectorQueries(ctx, clientId)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkGetCollectorByClientID(b *testing.B) {
|
||||
ctx := b.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(b, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
clientId := "EXAMPLE"
|
||||
err := queries.CreateClient(ctx, &repository.CreateClientParams{
|
||||
Name: "example_client",
|
||||
Clientid: clientId,
|
||||
})
|
||||
require.NoError(b, err)
|
||||
|
||||
b.ResetTimer()
|
||||
for b.Loop() {
|
||||
_, _ = queries.GetCollectorByClientID(ctx, clientId)
|
||||
}
|
||||
}
|
||||
@@ -167,74 +167,32 @@ func (q *Queries) GetDocumentEntry(ctx context.Context, documentid uuid.UUID) (*
|
||||
}
|
||||
|
||||
const getDocumentExternal = `-- name: GetDocumentExternal :one
|
||||
WITH
|
||||
docs AS (
|
||||
SELECT id, hash, clientId
|
||||
FROM documents
|
||||
WHERE id = $1
|
||||
),
|
||||
namedResults AS (
|
||||
SELECT
|
||||
q.name,
|
||||
dd.id,
|
||||
d.value
|
||||
FROM currentCollectorQueries AS q
|
||||
JOIN docs as dd on dd.clientId = q.clientId
|
||||
LEFT JOIN listValidDocumentResults($1) AS d
|
||||
ON d.queryId = q.queryId and q.name is not null
|
||||
)
|
||||
SELECT
|
||||
d.id,
|
||||
d.clientId,
|
||||
d.hash,
|
||||
COALESCE(jsonb_object_agg(r.name, r.value) FILTER (WHERE r.name IS NOT NULL), '{}') AS fields
|
||||
FROM docs AS d
|
||||
LEFT JOIN namedResults AS r ON d.id = r.id
|
||||
GROUP BY d.id, d.clientId, d.hash
|
||||
id,
|
||||
clientId,
|
||||
hash
|
||||
FROM documents
|
||||
WHERE id = $1
|
||||
`
|
||||
|
||||
type GetDocumentExternalRow struct {
|
||||
ID uuid.UUID `db:"id"`
|
||||
Clientid string `db:"clientid"`
|
||||
Hash string `db:"hash"`
|
||||
Fields []byte `db:"fields"`
|
||||
}
|
||||
|
||||
// GetDocumentExternal
|
||||
// Query functionality has been removed. See remove_query_plan.md for details.
|
||||
//
|
||||
// WITH
|
||||
// docs AS (
|
||||
// SELECT id, hash, clientId
|
||||
// FROM documents
|
||||
// WHERE id = $1
|
||||
// ),
|
||||
// namedResults AS (
|
||||
// SELECT
|
||||
// q.name,
|
||||
// dd.id,
|
||||
// d.value
|
||||
// FROM currentCollectorQueries AS q
|
||||
// JOIN docs as dd on dd.clientId = q.clientId
|
||||
// LEFT JOIN listValidDocumentResults($1) AS d
|
||||
// ON d.queryId = q.queryId and q.name is not null
|
||||
// )
|
||||
// SELECT
|
||||
// d.id,
|
||||
// d.clientId,
|
||||
// d.hash,
|
||||
// COALESCE(jsonb_object_agg(r.name, r.value) FILTER (WHERE r.name IS NOT NULL), '{}') AS fields
|
||||
// FROM docs AS d
|
||||
// LEFT JOIN namedResults AS r ON d.id = r.id
|
||||
// GROUP BY d.id, d.clientId, d.hash
|
||||
// id,
|
||||
// clientId,
|
||||
// hash
|
||||
// FROM documents
|
||||
// WHERE id = $1
|
||||
func (q *Queries) GetDocumentExternal(ctx context.Context, documentid uuid.UUID) (*GetDocumentExternalRow, error) {
|
||||
row := q.db.QueryRow(ctx, getDocumentExternal, documentid)
|
||||
var i GetDocumentExternalRow
|
||||
err := row.Scan(
|
||||
&i.ID,
|
||||
&i.Clientid,
|
||||
&i.Hash,
|
||||
&i.Fields,
|
||||
)
|
||||
err := row.Scan(&i.ID, &i.Clientid, &i.Hash)
|
||||
return &i, err
|
||||
}
|
||||
|
||||
|
||||
@@ -218,7 +218,6 @@ func TestDocument(t *testing.T) {
|
||||
ID: id,
|
||||
Clientid: clientId,
|
||||
Hash: hash,
|
||||
Fields: []byte("{}"),
|
||||
}, docext)
|
||||
})
|
||||
t.Run("doc id by hash", func(t *testing.T) {
|
||||
|
||||
@@ -181,57 +181,6 @@ func (e Cleanmimetype) Valid() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
type Querytype string
|
||||
|
||||
const (
|
||||
QuerytypeContextFull Querytype = "context_full"
|
||||
QuerytypeJsonExtractor Querytype = "json_extractor"
|
||||
)
|
||||
|
||||
func (e *Querytype) Scan(src interface{}) error {
|
||||
switch s := src.(type) {
|
||||
case []byte:
|
||||
*e = Querytype(s)
|
||||
case string:
|
||||
*e = Querytype(s)
|
||||
default:
|
||||
return fmt.Errorf("unsupported scan type for Querytype: %T", src)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type NullQuerytype struct {
|
||||
Querytype Querytype
|
||||
Valid bool // Valid is true if Querytype is not NULL
|
||||
}
|
||||
|
||||
// Scan implements the Scanner interface.
|
||||
func (ns *NullQuerytype) Scan(value interface{}) error {
|
||||
if value == nil {
|
||||
ns.Querytype, ns.Valid = "", false
|
||||
return nil
|
||||
}
|
||||
ns.Valid = true
|
||||
return ns.Querytype.Scan(value)
|
||||
}
|
||||
|
||||
// Value implements the driver Valuer interface.
|
||||
func (ns NullQuerytype) Value() (driver.Value, error) {
|
||||
if !ns.Valid {
|
||||
return nil, nil
|
||||
}
|
||||
return string(ns.Querytype), nil
|
||||
}
|
||||
|
||||
func (e Querytype) Valid() bool {
|
||||
switch e {
|
||||
case QuerytypeContextFull,
|
||||
QuerytypeJsonExtractor:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
type BatchUpload struct {
|
||||
ID uuid.UUID `db:"id"`
|
||||
ClientID string `db:"client_id"`
|
||||
@@ -293,23 +242,6 @@ type Collectormintextversion struct {
|
||||
Removedversion *int32 `db:"removedversion"`
|
||||
}
|
||||
|
||||
type Collectorquery struct {
|
||||
ID uuid.UUID `db:"id"`
|
||||
Clientid string `db:"clientid"`
|
||||
Name string `db:"name"`
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Addedversion int32 `db:"addedversion"`
|
||||
Removedversion *int32 `db:"removedversion"`
|
||||
}
|
||||
|
||||
type Collectorquerydependencytree struct {
|
||||
Clientid string `db:"clientid"`
|
||||
Queryid *uuid.UUID `db:"queryid"`
|
||||
Querytype Querytype `db:"querytype"`
|
||||
Queryversion int32 `db:"queryversion"`
|
||||
Requiredids []uuid.UUID `db:"requiredids"`
|
||||
}
|
||||
|
||||
type Collectorversion struct {
|
||||
Clientid string `db:"clientid"`
|
||||
ID int32 `db:"id"`
|
||||
@@ -343,17 +275,6 @@ type Currentcollectormintextversion struct {
|
||||
Mintextversion int64 `db:"mintextversion"`
|
||||
}
|
||||
|
||||
type Currentcollectorqueriesjsonagg struct {
|
||||
Clientid string `db:"clientid"`
|
||||
Fields []byte `db:"fields"`
|
||||
}
|
||||
|
||||
type Currentcollectorquery struct {
|
||||
Clientid string `db:"clientid"`
|
||||
Name *string `db:"name"`
|
||||
Queryid *uuid.UUID `db:"queryid"`
|
||||
}
|
||||
|
||||
type Currentfieldextraction struct {
|
||||
ID uuid.UUID `db:"id"`
|
||||
Documentid uuid.UUID `db:"documentid"`
|
||||
@@ -659,16 +580,6 @@ type Fullactivecollector struct {
|
||||
Mintextversion int64 `db:"mintextversion"`
|
||||
Activeversion int32 `db:"activeversion"`
|
||||
Latestversion int32 `db:"latestversion"`
|
||||
Fields []byte `db:"fields"`
|
||||
}
|
||||
|
||||
type Fullactivequery struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Querytype Querytype `db:"querytype"`
|
||||
Activeversion int32 `db:"activeversion"`
|
||||
Latestversion int32 `db:"latestversion"`
|
||||
Config []byte `db:"config"`
|
||||
Requiredids []uuid.UUID `db:"requiredids"`
|
||||
}
|
||||
|
||||
type Fullclient struct {
|
||||
@@ -681,79 +592,3 @@ type Label struct {
|
||||
Label string `db:"label"`
|
||||
Description string `db:"description"`
|
||||
}
|
||||
|
||||
type Query struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Querytype Querytype `db:"querytype"`
|
||||
}
|
||||
|
||||
type Queryactivedependency struct {
|
||||
ID uuid.UUID `db:"id"`
|
||||
Requiredqueryid *uuid.UUID `db:"requiredqueryid"`
|
||||
}
|
||||
|
||||
type Queryactiveversion struct {
|
||||
Activeversionentryid uuid.UUID `db:"activeversionentryid"`
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Versionid int32 `db:"versionid"`
|
||||
}
|
||||
|
||||
type Queryconfig struct {
|
||||
Configid uuid.UUID `db:"configid"`
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Config []byte `db:"config"`
|
||||
Addedversion int32 `db:"addedversion"`
|
||||
Removedversion *int32 `db:"removedversion"`
|
||||
}
|
||||
|
||||
type Querycurrentactiveversion struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Activeversion int32 `db:"activeversion"`
|
||||
}
|
||||
|
||||
type Querycurrentconfig struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Config []byte `db:"config"`
|
||||
}
|
||||
|
||||
type Querycurrentrequiredid struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Requiredqueryid *uuid.UUID `db:"requiredqueryid"`
|
||||
}
|
||||
|
||||
type Querycurrentrequiredidsagg struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Requiredids []uuid.UUID `db:"requiredids"`
|
||||
}
|
||||
|
||||
type Querylatestversion struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Latestversion int32 `db:"latestversion"`
|
||||
}
|
||||
|
||||
type Queryversion struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Versionid int32 `db:"versionid"`
|
||||
Addedat pgtype.Timestamp `db:"addedat"`
|
||||
}
|
||||
|
||||
type Requiredquery struct {
|
||||
Requiredqueryentryid uuid.UUID `db:"requiredqueryentryid"`
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Requiredqueryid uuid.UUID `db:"requiredqueryid"`
|
||||
Addedversion int32 `db:"addedversion"`
|
||||
Removedversion *int32 `db:"removedversion"`
|
||||
}
|
||||
|
||||
type Result struct {
|
||||
ID uuid.UUID `db:"id"`
|
||||
Textentryid uuid.UUID `db:"textentryid"`
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Value string `db:"value"`
|
||||
Queryversion int32 `db:"queryversion"`
|
||||
}
|
||||
|
||||
type Resultdependency struct {
|
||||
Resultid uuid.UUID `db:"resultid"`
|
||||
Requiredresultid uuid.UUID `db:"requiredresultid"`
|
||||
}
|
||||
|
||||
@@ -1,425 +0,0 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.27.0
|
||||
// source: query.sql
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
const addActiveQueryVersion = `-- name: AddActiveQueryVersion :exec
|
||||
INSERT INTO queryActiveVersions (queryId, versionId) VALUES ($1, $2)
|
||||
`
|
||||
|
||||
type AddActiveQueryVersionParams struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Versionid int32 `db:"versionid"`
|
||||
}
|
||||
|
||||
// AddActiveQueryVersion
|
||||
//
|
||||
// INSERT INTO queryActiveVersions (queryId, versionId) VALUES ($1, $2)
|
||||
func (q *Queries) AddActiveQueryVersion(ctx context.Context, arg *AddActiveQueryVersionParams) error {
|
||||
_, err := q.db.Exec(ctx, addActiveQueryVersion, arg.Queryid, arg.Versionid)
|
||||
return err
|
||||
}
|
||||
|
||||
const addLatestQueryVersion = `-- name: AddLatestQueryVersion :one
|
||||
INSERT INTO queryVersions (queryId) VALUES ($1) RETURNING versionId
|
||||
`
|
||||
|
||||
// AddLatestQueryVersion
|
||||
//
|
||||
// INSERT INTO queryVersions (queryId) VALUES ($1) RETURNING versionId
|
||||
func (q *Queries) AddLatestQueryVersion(ctx context.Context, queryid uuid.UUID) (int32, error) {
|
||||
row := q.db.QueryRow(ctx, addLatestQueryVersion, queryid)
|
||||
var versionid int32
|
||||
err := row.Scan(&versionid)
|
||||
return versionid, err
|
||||
}
|
||||
|
||||
const addRequiredQuery = `-- name: AddRequiredQuery :exec
|
||||
INSERT INTO requiredQueries (queryId, requiredQueryId, addedVersion) VALUES ($1, $2, $3)
|
||||
`
|
||||
|
||||
type AddRequiredQueryParams struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Requiredqueryid uuid.UUID `db:"requiredqueryid"`
|
||||
Addedversion int32 `db:"addedversion"`
|
||||
}
|
||||
|
||||
// AddRequiredQuery
|
||||
//
|
||||
// INSERT INTO requiredQueries (queryId, requiredQueryId, addedVersion) VALUES ($1, $2, $3)
|
||||
func (q *Queries) AddRequiredQuery(ctx context.Context, arg *AddRequiredQueryParams) error {
|
||||
_, err := q.db.Exec(ctx, addRequiredQuery, arg.Queryid, arg.Requiredqueryid, arg.Addedversion)
|
||||
return err
|
||||
}
|
||||
|
||||
const allQueriesExist = `-- name: AllQueriesExist :one
|
||||
SELECT COUNT(*) = COUNT(DISTINCT queryId) AS all_exist
|
||||
FROM unnest($1::uuid[]) AS input_id
|
||||
LEFT JOIN queries ON input_id = queries.queryId
|
||||
`
|
||||
|
||||
// AllQueriesExist
|
||||
//
|
||||
// SELECT COUNT(*) = COUNT(DISTINCT queryId) AS all_exist
|
||||
// FROM unnest($1::uuid[]) AS input_id
|
||||
// LEFT JOIN queries ON input_id = queries.queryId
|
||||
func (q *Queries) AllQueriesExist(ctx context.Context, dollar_1 []uuid.UUID) (bool, error) {
|
||||
row := q.db.QueryRow(ctx, allQueriesExist, dollar_1)
|
||||
var all_exist bool
|
||||
err := row.Scan(&all_exist)
|
||||
return all_exist, err
|
||||
}
|
||||
|
||||
const createQuery = `-- name: CreateQuery :one
|
||||
INSERT INTO queries (queryType) VALUES ($1) RETURNING queryId
|
||||
`
|
||||
|
||||
// CreateQuery
|
||||
//
|
||||
// INSERT INTO queries (queryType) VALUES ($1) RETURNING queryId
|
||||
func (q *Queries) CreateQuery(ctx context.Context, querytype Querytype) (uuid.UUID, error) {
|
||||
row := q.db.QueryRow(ctx, createQuery, querytype)
|
||||
var queryid uuid.UUID
|
||||
err := row.Scan(&queryid)
|
||||
return queryid, err
|
||||
}
|
||||
|
||||
const getActiveQueryConfig = `-- name: GetActiveQueryConfig :one
|
||||
SELECT config FROM queryCurrentConfigs where queryId = $1
|
||||
`
|
||||
|
||||
// GetActiveQueryConfig
|
||||
//
|
||||
// SELECT config FROM queryCurrentConfigs where queryId = $1
|
||||
func (q *Queries) GetActiveQueryConfig(ctx context.Context, queryid uuid.UUID) ([]byte, error) {
|
||||
row := q.db.QueryRow(ctx, getActiveQueryConfig, queryid)
|
||||
var config []byte
|
||||
err := row.Scan(&config)
|
||||
return config, err
|
||||
}
|
||||
|
||||
const getQuery = `-- name: GetQuery :one
|
||||
SELECT queryid, querytype, activeversion, latestversion, config, requiredids FROM fullActiveQueries WHERE queryId = $1
|
||||
`
|
||||
|
||||
// GetQuery
|
||||
//
|
||||
// SELECT queryid, querytype, activeversion, latestversion, config, requiredids FROM fullActiveQueries WHERE queryId = $1
|
||||
func (q *Queries) GetQuery(ctx context.Context, queryid uuid.UUID) (*Fullactivequery, error) {
|
||||
row := q.db.QueryRow(ctx, getQuery, queryid)
|
||||
var i Fullactivequery
|
||||
err := row.Scan(
|
||||
&i.Queryid,
|
||||
&i.Querytype,
|
||||
&i.Activeversion,
|
||||
&i.Latestversion,
|
||||
&i.Config,
|
||||
&i.Requiredids,
|
||||
)
|
||||
return &i, err
|
||||
}
|
||||
|
||||
const getQueryWithVersion = `-- name: GetQueryWithVersion :one
|
||||
WITH query as (
|
||||
SELECT queryId, queryType FROM queries WHERE queryId = $1
|
||||
),
|
||||
config as (
|
||||
SELECT c.queryId, c.config
|
||||
FROM query AS q
|
||||
LEFT JOIN queryConfigs AS c ON q.queryId = c.queryId
|
||||
and isInVersion($2, c.addedVersion, c.removedVersion)
|
||||
),
|
||||
requiredIds as (
|
||||
SELECT r.queryId,
|
||||
coalesce(
|
||||
ARRAY_AGG(DISTINCT r.requiredQueryId)
|
||||
FILTER (WHERE r.requiredQueryId != '00000000-0000-0000-0000-000000000000')::uuid[],
|
||||
array[]::uuid[]
|
||||
)::uuid[] as requiredIds
|
||||
FROM query AS q
|
||||
LEFT JOIN requiredQueries AS r ON q.queryId = r.queryId
|
||||
and isInVersion($2, r.addedVersion, r.removedVersion)
|
||||
GROUP BY r.queryId
|
||||
)
|
||||
SELECT DISTINCT q.queryId, q.queryType,
|
||||
av.activeVersion,
|
||||
lv.latestVersion, c.config,
|
||||
r.requiredIds
|
||||
FROM query AS q
|
||||
JOIN queryCurrentActiveVersions as av on q.queryId = av.queryId
|
||||
JOIN queryLatestVersions as lv on lv.queryId = q.queryId
|
||||
LEFT JOIN config AS c ON q.queryId = c.queryId
|
||||
LEFT JOIN requiredIds AS r ON q.queryId = r.queryId
|
||||
`
|
||||
|
||||
type GetQueryWithVersionParams struct {
|
||||
ID *uuid.UUID `db:"id"`
|
||||
Version *int32 `db:"version"`
|
||||
}
|
||||
|
||||
type GetQueryWithVersionRow struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Querytype Querytype `db:"querytype"`
|
||||
Activeversion int32 `db:"activeversion"`
|
||||
Latestversion int32 `db:"latestversion"`
|
||||
Config []byte `db:"config"`
|
||||
Requiredids []uuid.UUID `db:"requiredids"`
|
||||
}
|
||||
|
||||
// GetQueryWithVersion
|
||||
//
|
||||
// WITH query as (
|
||||
// SELECT queryId, queryType FROM queries WHERE queryId = $1
|
||||
// ),
|
||||
// config as (
|
||||
// SELECT c.queryId, c.config
|
||||
// FROM query AS q
|
||||
// LEFT JOIN queryConfigs AS c ON q.queryId = c.queryId
|
||||
// and isInVersion($2, c.addedVersion, c.removedVersion)
|
||||
// ),
|
||||
// requiredIds as (
|
||||
// SELECT r.queryId,
|
||||
// coalesce(
|
||||
// ARRAY_AGG(DISTINCT r.requiredQueryId)
|
||||
// FILTER (WHERE r.requiredQueryId != '00000000-0000-0000-0000-000000000000')::uuid[],
|
||||
// array[]::uuid[]
|
||||
// )::uuid[] as requiredIds
|
||||
// FROM query AS q
|
||||
// LEFT JOIN requiredQueries AS r ON q.queryId = r.queryId
|
||||
// and isInVersion($2, r.addedVersion, r.removedVersion)
|
||||
// GROUP BY r.queryId
|
||||
// )
|
||||
// SELECT DISTINCT q.queryId, q.queryType,
|
||||
// av.activeVersion,
|
||||
// lv.latestVersion, c.config,
|
||||
// r.requiredIds
|
||||
// FROM query AS q
|
||||
// JOIN queryCurrentActiveVersions as av on q.queryId = av.queryId
|
||||
// JOIN queryLatestVersions as lv on lv.queryId = q.queryId
|
||||
// LEFT JOIN config AS c ON q.queryId = c.queryId
|
||||
// LEFT JOIN requiredIds AS r ON q.queryId = r.queryId
|
||||
func (q *Queries) GetQueryWithVersion(ctx context.Context, arg *GetQueryWithVersionParams) (*GetQueryWithVersionRow, error) {
|
||||
row := q.db.QueryRow(ctx, getQueryWithVersion, arg.ID, arg.Version)
|
||||
var i GetQueryWithVersionRow
|
||||
err := row.Scan(
|
||||
&i.Queryid,
|
||||
&i.Querytype,
|
||||
&i.Activeversion,
|
||||
&i.Latestversion,
|
||||
&i.Config,
|
||||
&i.Requiredids,
|
||||
)
|
||||
return &i, err
|
||||
}
|
||||
|
||||
const isQueryInDependencyTree = `-- name: IsQueryInDependencyTree :one
|
||||
SELECT EXISTS (
|
||||
SELECT 1 FROM queryActiveDependencies
|
||||
WHERE id = any($1)
|
||||
and requiredQueryId = $2
|
||||
or $2 = any($1)
|
||||
)
|
||||
`
|
||||
|
||||
type IsQueryInDependencyTreeParams struct {
|
||||
Requiredqueryids []uuid.UUID `db:"requiredqueryids"`
|
||||
Queryid *uuid.UUID `db:"queryid"`
|
||||
}
|
||||
|
||||
// IsQueryInDependencyTree
|
||||
//
|
||||
// SELECT EXISTS (
|
||||
// SELECT 1 FROM queryActiveDependencies
|
||||
// WHERE id = any($1)
|
||||
// and requiredQueryId = $2
|
||||
// or $2 = any($1)
|
||||
// )
|
||||
func (q *Queries) IsQueryInDependencyTree(ctx context.Context, arg *IsQueryInDependencyTreeParams) (bool, error) {
|
||||
row := q.db.QueryRow(ctx, isQueryInDependencyTree, arg.Requiredqueryids, arg.Queryid)
|
||||
var exists bool
|
||||
err := row.Scan(&exists)
|
||||
return exists, err
|
||||
}
|
||||
|
||||
const listQueries = `-- name: ListQueries :many
|
||||
SELECT queryid, querytype, activeversion, latestversion, config, requiredids FROM fullActiveQueries
|
||||
`
|
||||
|
||||
// ListQueries
|
||||
//
|
||||
// SELECT queryid, querytype, 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()
|
||||
items := []*Fullactivequery{}
|
||||
for rows.Next() {
|
||||
var i Fullactivequery
|
||||
if err := rows.Scan(
|
||||
&i.Queryid,
|
||||
&i.Querytype,
|
||||
&i.Activeversion,
|
||||
&i.Latestversion,
|
||||
&i.Config,
|
||||
&i.Requiredids,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, &i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const listQueriesById = `-- name: ListQueriesById :many
|
||||
SELECT queryid, querytype, activeversion, latestversion, config, requiredids FROM fullActiveQueries WHERE queryId = any($1)
|
||||
`
|
||||
|
||||
// ListQueriesById
|
||||
//
|
||||
// SELECT queryid, querytype, activeversion, latestversion, config, requiredids FROM fullActiveQueries WHERE queryId = any($1)
|
||||
func (q *Queries) ListQueriesById(ctx context.Context, queryid []uuid.UUID) ([]*Fullactivequery, error) {
|
||||
rows, err := q.db.Query(ctx, listQueriesById, queryid)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []*Fullactivequery{}
|
||||
for rows.Next() {
|
||||
var i Fullactivequery
|
||||
if err := rows.Scan(
|
||||
&i.Queryid,
|
||||
&i.Querytype,
|
||||
&i.Activeversion,
|
||||
&i.Latestversion,
|
||||
&i.Config,
|
||||
&i.Requiredids,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, &i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const listQueryClientIDs = `-- name: ListQueryClientIDs :many
|
||||
SELECT clientId FROM collectorQueryDependencyTree WHERE queryId = $1
|
||||
`
|
||||
|
||||
// ListQueryClientIDs
|
||||
//
|
||||
// SELECT clientId FROM collectorQueryDependencyTree WHERE queryId = $1
|
||||
func (q *Queries) ListQueryClientIDs(ctx context.Context, queryid *uuid.UUID) ([]string, error) {
|
||||
rows, err := q.db.Query(ctx, listQueryClientIDs, queryid)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []string{}
|
||||
for rows.Next() {
|
||||
var clientid string
|
||||
if err := rows.Scan(&clientid); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, clientid)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const listQueryDirectDependentsByDocumentID = `-- name: ListQueryDirectDependentsByDocumentID :many
|
||||
WITH doc AS (
|
||||
SELECT clientId FROM documents WHERE id = $2
|
||||
)
|
||||
SELECT dt.queryId
|
||||
FROM doc,
|
||||
LATERAL collectorQueryDependencyTreeByClient(doc.clientId) AS dt
|
||||
WHERE $1 = ANY(dt.requiredIds)
|
||||
`
|
||||
|
||||
type ListQueryDirectDependentsByDocumentIDParams struct {
|
||||
Queryid *uuid.UUID `db:"queryid"`
|
||||
Documentid *uuid.UUID `db:"documentid"`
|
||||
}
|
||||
|
||||
// ListQueryDirectDependentsByDocumentID
|
||||
//
|
||||
// WITH doc AS (
|
||||
// SELECT clientId FROM documents WHERE id = $2
|
||||
// )
|
||||
// SELECT dt.queryId
|
||||
// FROM doc,
|
||||
// LATERAL collectorQueryDependencyTreeByClient(doc.clientId) AS dt
|
||||
// WHERE $1 = ANY(dt.requiredIds)
|
||||
func (q *Queries) ListQueryDirectDependentsByDocumentID(ctx context.Context, arg *ListQueryDirectDependentsByDocumentIDParams) ([]*uuid.UUID, error) {
|
||||
rows, err := q.db.Query(ctx, listQueryDirectDependentsByDocumentID, arg.Queryid, arg.Documentid)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []*uuid.UUID{}
|
||||
for rows.Next() {
|
||||
var queryid *uuid.UUID
|
||||
if err := rows.Scan(&queryid); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, queryid)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const removeRequiredQuery = `-- name: RemoveRequiredQuery :exec
|
||||
UPDATE requiredQueries SET removedVersion = $1 WHERE requiredQueryId = $2 and queryId = $3 and removedVersion is null
|
||||
`
|
||||
|
||||
type RemoveRequiredQueryParams struct {
|
||||
Removedversion *int32 `db:"removedversion"`
|
||||
Requiredqueryid uuid.UUID `db:"requiredqueryid"`
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
}
|
||||
|
||||
// RemoveRequiredQuery
|
||||
//
|
||||
// UPDATE requiredQueries SET removedVersion = $1 WHERE requiredQueryId = $2 and queryId = $3 and removedVersion is null
|
||||
func (q *Queries) RemoveRequiredQuery(ctx context.Context, arg *RemoveRequiredQueryParams) error {
|
||||
_, err := q.db.Exec(ctx, removeRequiredQuery, arg.Removedversion, arg.Requiredqueryid, arg.Queryid)
|
||||
return err
|
||||
}
|
||||
|
||||
const setQueryConfig = `-- name: SetQueryConfig :exec
|
||||
INSERT INTO queryConfigs (queryId, config, addedVersion) VALUES ($1, $2, $3)
|
||||
`
|
||||
|
||||
type SetQueryConfigParams struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Config []byte `db:"config"`
|
||||
Addedversion int32 `db:"addedversion"`
|
||||
}
|
||||
|
||||
// SetQueryConfig
|
||||
//
|
||||
// INSERT INTO queryConfigs (queryId, config, addedVersion) VALUES ($1, $2, $3)
|
||||
func (q *Queries) SetQueryConfig(ctx context.Context, arg *SetQueryConfigParams) error {
|
||||
_, err := q.db.Exec(ctx, setQueryConfig, arg.Queryid, arg.Config, arg.Addedversion)
|
||||
return err
|
||||
}
|
||||
@@ -1,622 +0,0 @@
|
||||
package repository_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"queryorchestration/internal/database/repository"
|
||||
"queryorchestration/internal/serviceconfig"
|
||||
"queryorchestration/internal/test"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestQueries(t *testing.T) {
|
||||
t.Parallel()
|
||||
if testing.Short() {
|
||||
t.SkipNow()
|
||||
}
|
||||
ctx := t.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
contextQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull))
|
||||
require.NoError(t, err)
|
||||
|
||||
contextQuery, err := queries.GetQuery(ctx, contextQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivequery{
|
||||
Queryid: contextQueryID,
|
||||
Querytype: repository.QuerytypeContextFull,
|
||||
Activeversion: 0,
|
||||
Latestversion: 0,
|
||||
Config: nil,
|
||||
Requiredids: []uuid.UUID{},
|
||||
}, contextQuery)
|
||||
|
||||
ctxVersion, err := queries.AddLatestQueryVersion(ctx, contextQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, int32(1), ctxVersion)
|
||||
|
||||
contextQuery, err = queries.GetQuery(ctx, contextQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivequery{
|
||||
Queryid: contextQueryID,
|
||||
Querytype: repository.QuerytypeContextFull,
|
||||
Activeversion: 0,
|
||||
Latestversion: 1,
|
||||
Config: nil,
|
||||
Requiredids: []uuid.UUID{},
|
||||
}, contextQuery)
|
||||
|
||||
jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor))
|
||||
require.NoError(t, err)
|
||||
|
||||
jsonQuery, err := queries.GetQuery(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivequery{
|
||||
Queryid: jsonQueryID,
|
||||
Querytype: repository.QuerytypeJsonExtractor,
|
||||
Activeversion: 0,
|
||||
Latestversion: 0,
|
||||
Config: nil,
|
||||
Requiredids: []uuid.UUID{},
|
||||
}, jsonQuery)
|
||||
|
||||
jsonVersion, err := queries.AddLatestQueryVersion(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, int32(1), jsonVersion)
|
||||
|
||||
jsonQuery, err = queries.GetQuery(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivequery{
|
||||
Queryid: jsonQueryID,
|
||||
Querytype: repository.QuerytypeJsonExtractor,
|
||||
Activeversion: 0,
|
||||
Latestversion: 1,
|
||||
Config: nil,
|
||||
Requiredids: []uuid.UUID{},
|
||||
}, jsonQuery)
|
||||
|
||||
err = queries.AddActiveQueryVersion(ctx, &repository.AddActiveQueryVersionParams{
|
||||
Versionid: 1,
|
||||
Queryid: jsonQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
jsonQuery, err = queries.GetQuery(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivequery{
|
||||
Queryid: jsonQueryID,
|
||||
Querytype: repository.QuerytypeJsonExtractor,
|
||||
Activeversion: 1,
|
||||
Latestversion: 1,
|
||||
Config: nil,
|
||||
Requiredids: []uuid.UUID{},
|
||||
}, jsonQuery)
|
||||
|
||||
jsonConfig := []byte("{\"path\": \"example_path\"}")
|
||||
|
||||
err = queries.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
|
||||
Queryid: jsonQueryID,
|
||||
Requiredqueryid: contextQueryID,
|
||||
Addedversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
jsonQuery, err = queries.GetQuery(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivequery{
|
||||
Queryid: jsonQueryID,
|
||||
Querytype: repository.QuerytypeJsonExtractor,
|
||||
Activeversion: 1,
|
||||
Latestversion: 1,
|
||||
Config: nil,
|
||||
Requiredids: []uuid.UUID{contextQueryID},
|
||||
}, jsonQuery)
|
||||
|
||||
jsonVersion, err = queries.AddLatestQueryVersion(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, int32(2), jsonVersion)
|
||||
|
||||
jsonQuery, err = queries.GetQuery(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivequery{
|
||||
Queryid: jsonQueryID,
|
||||
Querytype: repository.QuerytypeJsonExtractor,
|
||||
Activeversion: 1,
|
||||
Latestversion: 2,
|
||||
Config: nil,
|
||||
Requiredids: []uuid.UUID{contextQueryID},
|
||||
}, jsonQuery)
|
||||
|
||||
removeV := int32(2)
|
||||
err = queries.RemoveRequiredQuery(ctx, &repository.RemoveRequiredQueryParams{
|
||||
Queryid: jsonQueryID,
|
||||
Requiredqueryid: contextQueryID,
|
||||
Removedversion: &removeV,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
err = queries.SetQueryConfig(ctx, &repository.SetQueryConfigParams{
|
||||
Queryid: jsonQueryID,
|
||||
Config: jsonConfig,
|
||||
Addedversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
jsonQueryConfig, err := queries.GetActiveQueryConfig(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, jsonConfig, jsonQueryConfig)
|
||||
|
||||
err = queries.SetQueryConfig(ctx, &repository.SetQueryConfigParams{
|
||||
Queryid: jsonQueryID,
|
||||
Config: []byte(`{"second":"key"}`),
|
||||
Addedversion: 2,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
jsonQueryConfig, err = queries.GetActiveQueryConfig(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, jsonConfig, jsonQueryConfig)
|
||||
|
||||
err = queries.AddActiveQueryVersion(ctx, &repository.AddActiveQueryVersionParams{
|
||||
Queryid: jsonQueryID,
|
||||
Versionid: 2,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
jsonQueryConfig, err = queries.GetActiveQueryConfig(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, []byte(`{"second": "key"}`), jsonQueryConfig)
|
||||
|
||||
jsonQuery, err = queries.GetQuery(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.Fullactivequery{
|
||||
Queryid: jsonQueryID,
|
||||
Querytype: repository.QuerytypeJsonExtractor,
|
||||
Activeversion: 2,
|
||||
Latestversion: 2,
|
||||
Config: jsonQueryConfig,
|
||||
Requiredids: []uuid.UUID{},
|
||||
}, jsonQuery)
|
||||
|
||||
v := int32(1)
|
||||
versionedQuery, err := queries.GetQueryWithVersion(ctx, &repository.GetQueryWithVersionParams{
|
||||
ID: &jsonQueryID,
|
||||
Version: &v,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.EqualExportedValues(t, &repository.GetQueryWithVersionRow{
|
||||
Queryid: jsonQueryID,
|
||||
Querytype: repository.QuerytypeJsonExtractor,
|
||||
Activeversion: 2,
|
||||
Latestversion: 2,
|
||||
Config: jsonConfig,
|
||||
Requiredids: []uuid.UUID{contextQueryID},
|
||||
}, versionedQuery)
|
||||
|
||||
all_exist, err := queries.AllQueriesExist(ctx, []uuid.UUID{})
|
||||
require.NoError(t, err)
|
||||
assert.True(t, all_exist)
|
||||
|
||||
all_exist, err = queries.AllQueriesExist(ctx, []uuid.UUID{uuid.New()})
|
||||
require.NoError(t, err)
|
||||
assert.False(t, all_exist)
|
||||
|
||||
all_exist, err = queries.AllQueriesExist(ctx, []uuid.UUID{jsonQueryID})
|
||||
require.NoError(t, err)
|
||||
assert.True(t, all_exist)
|
||||
|
||||
all_exist, err = queries.AllQueriesExist(ctx, []uuid.UUID{jsonQueryID, contextQueryID})
|
||||
require.NoError(t, err)
|
||||
assert.True(t, all_exist)
|
||||
|
||||
all_exist, err = queries.AllQueriesExist(ctx, []uuid.UUID{jsonQueryID, uuid.New()})
|
||||
require.NoError(t, err)
|
||||
assert.False(t, all_exist)
|
||||
}
|
||||
|
||||
func TestQueryDependencyTree(t *testing.T) {
|
||||
t.Parallel()
|
||||
if testing.Short() {
|
||||
t.SkipNow()
|
||||
}
|
||||
ctx := t.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
clientID := "EXAMPLE"
|
||||
err := queries.CreateClient(ctx, &repository.CreateClientParams{
|
||||
Name: "example_client",
|
||||
Clientid: clientID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
version, err := queries.AddLatestCollectorVersion(ctx, clientID)
|
||||
require.NoError(t, err)
|
||||
err = queries.SetActiveCollectorVersion(ctx, &repository.SetActiveCollectorVersionParams{
|
||||
Clientid: clientID,
|
||||
Versionid: version,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
docID, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{
|
||||
Clientid: clientID,
|
||||
Hash: "sample",
|
||||
BatchID: nil,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
contextQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull))
|
||||
require.NoError(t, err)
|
||||
_, err = queries.AddLatestQueryVersion(ctx, contextQueryID)
|
||||
require.NoError(t, err)
|
||||
err = queries.AddActiveQueryVersion(ctx, &repository.AddActiveQueryVersionParams{
|
||||
Queryid: contextQueryID,
|
||||
Versionid: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
dependents, err := queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &contextQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []uuid.UUID{}, dependents)
|
||||
|
||||
jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor))
|
||||
require.NoError(t, err)
|
||||
_, err = queries.AddLatestQueryVersion(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
err = queries.AddActiveQueryVersion(ctx, &repository.AddActiveQueryVersionParams{
|
||||
Queryid: jsonQueryID,
|
||||
Versionid: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &jsonQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []uuid.UUID{}, dependents)
|
||||
|
||||
err = queries.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
|
||||
Queryid: jsonQueryID,
|
||||
Requiredqueryid: contextQueryID,
|
||||
Addedversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &jsonQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []uuid.UUID{}, dependents)
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &contextQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []uuid.UUID{}, dependents)
|
||||
|
||||
err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
|
||||
Clientid: clientID,
|
||||
Name: "example_name'",
|
||||
Queryid: jsonQueryID,
|
||||
Addedversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &jsonQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{}, dependents)
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &contextQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{&jsonQueryID}, dependents)
|
||||
|
||||
secondJsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor))
|
||||
require.NoError(t, err)
|
||||
_, err = queries.AddLatestQueryVersion(ctx, secondJsonQueryID)
|
||||
require.NoError(t, err)
|
||||
err = queries.AddActiveQueryVersion(ctx, &repository.AddActiveQueryVersionParams{
|
||||
Queryid: secondJsonQueryID,
|
||||
Versionid: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &jsonQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{}, dependents)
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &contextQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{&jsonQueryID}, dependents)
|
||||
|
||||
err = queries.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
|
||||
Queryid: secondJsonQueryID,
|
||||
Requiredqueryid: jsonQueryID,
|
||||
Addedversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &secondJsonQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{}, dependents)
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &jsonQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{}, dependents)
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &contextQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{&jsonQueryID}, dependents)
|
||||
|
||||
err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
|
||||
Clientid: clientID,
|
||||
Name: "example_name'",
|
||||
Queryid: secondJsonQueryID,
|
||||
Addedversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &secondJsonQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{}, dependents)
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &jsonQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{&secondJsonQueryID}, dependents)
|
||||
dependents, err = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &contextQueryID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{&jsonQueryID}, dependents)
|
||||
|
||||
isdependent, err := queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{
|
||||
Queryid: &jsonQueryID,
|
||||
Requiredqueryids: []uuid.UUID{contextQueryID},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.False(t, isdependent)
|
||||
|
||||
isdependent, err = queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{
|
||||
Queryid: &jsonQueryID,
|
||||
Requiredqueryids: []uuid.UUID{secondJsonQueryID},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.True(t, isdependent)
|
||||
|
||||
isdependent, err = queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{
|
||||
Queryid: &jsonQueryID,
|
||||
Requiredqueryids: []uuid.UUID{jsonQueryID},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.True(t, isdependent)
|
||||
|
||||
isdependent, err = queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{
|
||||
Queryid: &secondJsonQueryID,
|
||||
Requiredqueryids: []uuid.UUID{jsonQueryID, contextQueryID},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.False(t, isdependent)
|
||||
|
||||
isdependent, err = queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{
|
||||
Queryid: &contextQueryID,
|
||||
Requiredqueryids: []uuid.UUID{jsonQueryID, secondJsonQueryID},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.True(t, isdependent)
|
||||
}
|
||||
|
||||
func TestQueriesList(t *testing.T) {
|
||||
t.Parallel()
|
||||
if testing.Short() {
|
||||
t.SkipNow()
|
||||
}
|
||||
ctx := t.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
contextQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull))
|
||||
require.NoError(t, err)
|
||||
|
||||
jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor))
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err := queries.ListQueries(ctx)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 2)
|
||||
assert.ElementsMatch(t, []*repository.Fullactivequery{
|
||||
{
|
||||
Queryid: jsonQueryID,
|
||||
Querytype: repository.QuerytypeJsonExtractor,
|
||||
Activeversion: 0,
|
||||
Latestversion: 0,
|
||||
Config: nil,
|
||||
Requiredids: []uuid.UUID{},
|
||||
},
|
||||
{
|
||||
Queryid: contextQueryID,
|
||||
Querytype: repository.QuerytypeContextFull,
|
||||
Activeversion: 0,
|
||||
Latestversion: 0,
|
||||
Config: nil,
|
||||
Requiredids: []uuid.UUID{},
|
||||
},
|
||||
}, qs)
|
||||
|
||||
qs, err = queries.ListQueriesById(ctx, []uuid.UUID{jsonQueryID})
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 1)
|
||||
assert.ElementsMatch(t, []*repository.Fullactivequery{
|
||||
{
|
||||
Queryid: jsonQueryID,
|
||||
Querytype: repository.QuerytypeJsonExtractor,
|
||||
Activeversion: 0,
|
||||
Latestversion: 0,
|
||||
Config: nil,
|
||||
Requiredids: []uuid.UUID{},
|
||||
},
|
||||
}, qs)
|
||||
}
|
||||
|
||||
func TestListQueryClients(t *testing.T) {
|
||||
t.Parallel()
|
||||
if testing.Short() {
|
||||
t.SkipNow()
|
||||
}
|
||||
ctx := t.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
contextID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull))
|
||||
require.NoError(t, err)
|
||||
|
||||
clients, err := queries.ListQueryClientIDs(ctx, &contextID)
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []uuid.UUID{}, clients)
|
||||
|
||||
clientOneID := "EXAMPLE"
|
||||
err = queries.CreateClient(ctx, &repository.CreateClientParams{
|
||||
Name: "example_client",
|
||||
Clientid: clientOneID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
versionOne, err := queries.AddLatestCollectorVersion(ctx, clientOneID)
|
||||
require.NoError(t, err)
|
||||
err = queries.SetActiveCollectorVersion(ctx, &repository.SetActiveCollectorVersionParams{
|
||||
Clientid: clientOneID,
|
||||
Versionid: versionOne,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
|
||||
Clientid: clientOneID,
|
||||
Queryid: contextID,
|
||||
Addedversion: 1,
|
||||
Name: "example_key",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
clients, err = queries.ListQueryClientIDs(ctx, &contextID)
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []string{clientOneID}, clients)
|
||||
|
||||
clientTwoID := "EXAMPLE_DOS"
|
||||
err = queries.CreateClient(ctx, &repository.CreateClientParams{
|
||||
Name: "example_client_dos",
|
||||
Clientid: clientTwoID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
versionTwo, err := queries.AddLatestCollectorVersion(ctx, clientTwoID)
|
||||
require.NoError(t, err)
|
||||
err = queries.SetActiveCollectorVersion(ctx, &repository.SetActiveCollectorVersionParams{
|
||||
Clientid: clientTwoID,
|
||||
Versionid: versionTwo,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
|
||||
Clientid: clientTwoID,
|
||||
Queryid: contextID,
|
||||
Addedversion: 1,
|
||||
Name: "example_key",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
clients, err = queries.ListQueryClientIDs(ctx, &contextID)
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []string{clientOneID, clientTwoID}, clients)
|
||||
|
||||
jsonID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor))
|
||||
require.NoError(t, err)
|
||||
_, err = queries.AddLatestQueryVersion(ctx, jsonID)
|
||||
require.NoError(t, err)
|
||||
err = queries.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
|
||||
Queryid: jsonID,
|
||||
Requiredqueryid: contextID,
|
||||
Addedversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
|
||||
Clientid: clientOneID,
|
||||
Queryid: jsonID,
|
||||
Addedversion: 1,
|
||||
Name: "example_key",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
clients, err = queries.ListQueryClientIDs(ctx, &jsonID)
|
||||
require.NoError(t, err)
|
||||
assert.ElementsMatch(t, []string{clientOneID}, clients)
|
||||
}
|
||||
|
||||
func BenchmarkListQueryDirectDependentsByDocId(b *testing.B) {
|
||||
ctx := b.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(b, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
clientID := "EXAMPLE"
|
||||
err := queries.CreateClient(ctx, &repository.CreateClientParams{
|
||||
Name: "example_client",
|
||||
Clientid: clientID,
|
||||
})
|
||||
require.NoError(b, err)
|
||||
|
||||
docID, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{
|
||||
Clientid: clientID,
|
||||
Hash: "sample",
|
||||
BatchID: nil,
|
||||
})
|
||||
require.NoError(b, err)
|
||||
|
||||
contextQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull))
|
||||
require.NoError(b, err)
|
||||
|
||||
b.ResetTimer()
|
||||
for b.Loop() {
|
||||
_, _ = queries.ListQueryDirectDependentsByDocumentID(ctx, &repository.ListQueryDirectDependentsByDocumentIDParams{
|
||||
Documentid: &docID,
|
||||
Queryid: &contextQueryID,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,43 +0,0 @@
|
||||
package repository_test
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"queryorchestration/internal/database/repository"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestQueryTypeScan(t *testing.T) {
|
||||
qType := repository.Querytype(fmt.Sprint(0))
|
||||
|
||||
stringType := "context_full"
|
||||
err := qType.Scan(stringType)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestNullQueryTypeScan(t *testing.T) {
|
||||
qType := repository.NullQuerytype{}
|
||||
|
||||
stringType := "context_full"
|
||||
err := qType.Scan(stringType)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestNullQueryTypeValue(t *testing.T) {
|
||||
qType := repository.NullQuerytype{}
|
||||
|
||||
stringType := "context_full"
|
||||
err := qType.Scan(stringType)
|
||||
require.NoError(t, err)
|
||||
|
||||
val, err := qType.Value()
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, stringType, val)
|
||||
}
|
||||
|
||||
func TestQueryTypeValid(t *testing.T) {
|
||||
assert.True(t, repository.QuerytypeContextFull.Valid())
|
||||
}
|
||||
@@ -1,308 +0,0 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.27.0
|
||||
// source: result.sql
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
const addResult = `-- name: AddResult :one
|
||||
INSERT INTO results (queryId, value, textEntryId, queryVersion) VALUES ($1, $2, $3, $4) returning id
|
||||
`
|
||||
|
||||
type AddResultParams struct {
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Value string `db:"value"`
|
||||
Textentryid uuid.UUID `db:"textentryid"`
|
||||
Queryversion int32 `db:"queryversion"`
|
||||
}
|
||||
|
||||
// AddResult
|
||||
//
|
||||
// INSERT INTO results (queryId, value, textEntryId, queryVersion) VALUES ($1, $2, $3, $4) returning id
|
||||
func (q *Queries) AddResult(ctx context.Context, arg *AddResultParams) (uuid.UUID, error) {
|
||||
row := q.db.QueryRow(ctx, addResult,
|
||||
arg.Queryid,
|
||||
arg.Value,
|
||||
arg.Textentryid,
|
||||
arg.Queryversion,
|
||||
)
|
||||
var id uuid.UUID
|
||||
err := row.Scan(&id)
|
||||
return id, err
|
||||
}
|
||||
|
||||
const addResultDependency = `-- name: AddResultDependency :exec
|
||||
INSERT INTO resultDependencies (resultId, requiredResultId) VALUES ($1, $2)
|
||||
`
|
||||
|
||||
type AddResultDependencyParams struct {
|
||||
Resultid uuid.UUID `db:"resultid"`
|
||||
Requiredresultid uuid.UUID `db:"requiredresultid"`
|
||||
}
|
||||
|
||||
// AddResultDependency
|
||||
//
|
||||
// INSERT INTO resultDependencies (resultId, requiredResultId) VALUES ($1, $2)
|
||||
func (q *Queries) AddResultDependency(ctx context.Context, arg *AddResultDependencyParams) error {
|
||||
_, err := q.db.Exec(ctx, addResultDependency, arg.Resultid, arg.Requiredresultid)
|
||||
return err
|
||||
}
|
||||
|
||||
const getResultValueWithVersion = `-- name: GetResultValueWithVersion :one
|
||||
WITH doc as (
|
||||
SELECT id, clientId
|
||||
FROM documents
|
||||
WHERE id = $3
|
||||
)
|
||||
SELECT r.id, r.value
|
||||
FROM doc as d
|
||||
JOIN currentTextEntries as cte on cte.documentId = d.id
|
||||
LEFT JOIN results as r
|
||||
on r.queryId = $1
|
||||
and r.queryVersion = $2
|
||||
and r.textEntryId = cte.id
|
||||
`
|
||||
|
||||
type GetResultValueWithVersionParams struct {
|
||||
Queryid *uuid.UUID `db:"queryid"`
|
||||
Queryversion *int32 `db:"queryversion"`
|
||||
Documentid *uuid.UUID `db:"documentid"`
|
||||
}
|
||||
|
||||
type GetResultValueWithVersionRow struct {
|
||||
ID *uuid.UUID `db:"id"`
|
||||
Value *string `db:"value"`
|
||||
}
|
||||
|
||||
// GetResultValueWithVersion
|
||||
//
|
||||
// WITH doc as (
|
||||
// SELECT id, clientId
|
||||
// FROM documents
|
||||
// WHERE id = $3
|
||||
// )
|
||||
// SELECT r.id, r.value
|
||||
// FROM doc as d
|
||||
// JOIN currentTextEntries as cte on cte.documentId = d.id
|
||||
// LEFT JOIN results as r
|
||||
// on r.queryId = $1
|
||||
// and r.queryVersion = $2
|
||||
// and r.textEntryId = cte.id
|
||||
func (q *Queries) GetResultValueWithVersion(ctx context.Context, arg *GetResultValueWithVersionParams) (*GetResultValueWithVersionRow, error) {
|
||||
row := q.db.QueryRow(ctx, getResultValueWithVersion, arg.Queryid, arg.Queryversion, arg.Documentid)
|
||||
var i GetResultValueWithVersionRow
|
||||
err := row.Scan(&i.ID, &i.Value)
|
||||
return &i, err
|
||||
}
|
||||
|
||||
const listQueryRequirementValues = `-- name: ListQueryRequirementValues :many
|
||||
WITH reqQueries as (
|
||||
SELECT av.queryId, av.activeVersion, q.queryType
|
||||
FROM requiredQueries as rq
|
||||
JOIN queryCurrentActiveVersions as av on av.queryId = rq.requiredQueryId
|
||||
JOIN queries as q on q.queryId = av.queryId
|
||||
WHERE rq.queryId = $1
|
||||
and isInVersion($2, rq.addedVersion, rq.removedVersion)
|
||||
),
|
||||
docs as (
|
||||
SELECT id, clientId
|
||||
FROM documents
|
||||
WHERE id = $3
|
||||
),
|
||||
codeVersions as (
|
||||
SELECT
|
||||
d.id as documentId,
|
||||
mcv.minCleanVersion,
|
||||
mtv.minTextVersion
|
||||
FROM docs as d
|
||||
JOIN currentCollectorMinTextVersions as mtv on mtv.clientId = d.clientId
|
||||
JOIN currentCollectorMinCleanVersions as mcv on mcv.clientId = d.clientId
|
||||
),
|
||||
latestVersions AS (
|
||||
SELECT
|
||||
r.id,
|
||||
rq.queryId,
|
||||
rq.queryType,
|
||||
r.queryVersion,
|
||||
r.textEntryId,
|
||||
r.value,
|
||||
ROW_NUMBER() OVER (
|
||||
PARTITION BY r.queryId
|
||||
ORDER BY r.id DESC
|
||||
) as rowNumber
|
||||
FROM reqQueries as rq
|
||||
JOIN currentTextEntries as cte on cte.documentId = $3
|
||||
LEFT JOIN results as r ON rq.queryId = r.queryId
|
||||
and r.queryVersion = rq.activeVersion
|
||||
and cte.id = r.textEntryId
|
||||
)
|
||||
SELECT DISTINCT id, queryId, queryType, value
|
||||
FROM latestVersions
|
||||
WHERE rowNumber = 1
|
||||
`
|
||||
|
||||
type ListQueryRequirementValuesParams struct {
|
||||
Queryid *uuid.UUID `db:"queryid"`
|
||||
Version *int32 `db:"version"`
|
||||
Documentid *uuid.UUID `db:"documentid"`
|
||||
}
|
||||
|
||||
type ListQueryRequirementValuesRow struct {
|
||||
ID *uuid.UUID `db:"id"`
|
||||
Queryid uuid.UUID `db:"queryid"`
|
||||
Querytype Querytype `db:"querytype"`
|
||||
Value *string `db:"value"`
|
||||
}
|
||||
|
||||
// ListQueryRequirementValues
|
||||
//
|
||||
// WITH reqQueries as (
|
||||
// SELECT av.queryId, av.activeVersion, q.queryType
|
||||
// FROM requiredQueries as rq
|
||||
// JOIN queryCurrentActiveVersions as av on av.queryId = rq.requiredQueryId
|
||||
// JOIN queries as q on q.queryId = av.queryId
|
||||
// WHERE rq.queryId = $1
|
||||
// and isInVersion($2, rq.addedVersion, rq.removedVersion)
|
||||
// ),
|
||||
// docs as (
|
||||
// SELECT id, clientId
|
||||
// FROM documents
|
||||
// WHERE id = $3
|
||||
// ),
|
||||
// codeVersions as (
|
||||
// SELECT
|
||||
// d.id as documentId,
|
||||
// mcv.minCleanVersion,
|
||||
// mtv.minTextVersion
|
||||
// FROM docs as d
|
||||
// JOIN currentCollectorMinTextVersions as mtv on mtv.clientId = d.clientId
|
||||
// JOIN currentCollectorMinCleanVersions as mcv on mcv.clientId = d.clientId
|
||||
// ),
|
||||
// latestVersions AS (
|
||||
// SELECT
|
||||
// r.id,
|
||||
// rq.queryId,
|
||||
// rq.queryType,
|
||||
// r.queryVersion,
|
||||
// r.textEntryId,
|
||||
// r.value,
|
||||
// ROW_NUMBER() OVER (
|
||||
// PARTITION BY r.queryId
|
||||
// ORDER BY r.id DESC
|
||||
// ) as rowNumber
|
||||
// FROM reqQueries as rq
|
||||
// JOIN currentTextEntries as cte on cte.documentId = $3
|
||||
// LEFT JOIN results as r ON rq.queryId = r.queryId
|
||||
// and r.queryVersion = rq.activeVersion
|
||||
// and cte.id = r.textEntryId
|
||||
// )
|
||||
// SELECT DISTINCT id, queryId, queryType, value
|
||||
// FROM latestVersions
|
||||
// WHERE rowNumber = 1
|
||||
func (q *Queries) ListQueryRequirementValues(ctx context.Context, arg *ListQueryRequirementValuesParams) ([]*ListQueryRequirementValuesRow, error) {
|
||||
rows, err := q.db.Query(ctx, listQueryRequirementValues, arg.Queryid, arg.Version, arg.Documentid)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []*ListQueryRequirementValuesRow{}
|
||||
for rows.Next() {
|
||||
var i ListQueryRequirementValuesRow
|
||||
if err := rows.Scan(
|
||||
&i.ID,
|
||||
&i.Queryid,
|
||||
&i.Querytype,
|
||||
&i.Value,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, &i)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const listUnsyncedNoDepsQueriesByDocId = `-- name: ListUnsyncedNoDepsQueriesByDocId :many
|
||||
WITH docs as (
|
||||
SELECT id, clientId from documents where id = $1
|
||||
),
|
||||
dependency_tree as (
|
||||
select q.queryId, q.requiredIds, q.clientId, q.queryVersion
|
||||
from docs,
|
||||
LATERAL collectorQueryDependencyTreeByClient(docs.clientId) AS q
|
||||
),
|
||||
text_entries as (
|
||||
select id, documentId from currentTextEntries where documentId = $1
|
||||
),
|
||||
unsyncedQueries AS (
|
||||
SELECT DISTINCT dt.queryId, dt.requiredIds
|
||||
from docs as d
|
||||
JOIN dependency_tree as dt on d.clientId = dt.clientId
|
||||
JOIN text_entries as cte on cte.documentId = d.id
|
||||
LEFT JOIN results as r
|
||||
on r.queryId = dt.queryId
|
||||
and r.queryVersion = dt.queryVersion
|
||||
and cte.id = r.textEntryId
|
||||
where r.value is null
|
||||
)
|
||||
SELECT DISTINCT queryId FROM unsyncedQueries as baseuq
|
||||
WHERE NOT EXISTS (
|
||||
SELECT 1 FROM unsyncedQueries as uq WHERE uq.queryId = any(baseuq.requiredIds)
|
||||
)
|
||||
`
|
||||
|
||||
// ListUnsyncedNoDepsQueriesByDocId
|
||||
//
|
||||
// WITH docs as (
|
||||
// SELECT id, clientId from documents where id = $1
|
||||
// ),
|
||||
// dependency_tree as (
|
||||
// select q.queryId, q.requiredIds, q.clientId, q.queryVersion
|
||||
// from docs,
|
||||
// LATERAL collectorQueryDependencyTreeByClient(docs.clientId) AS q
|
||||
// ),
|
||||
// text_entries as (
|
||||
// select id, documentId from currentTextEntries where documentId = $1
|
||||
// ),
|
||||
// unsyncedQueries AS (
|
||||
// SELECT DISTINCT dt.queryId, dt.requiredIds
|
||||
// from docs as d
|
||||
// JOIN dependency_tree as dt on d.clientId = dt.clientId
|
||||
// JOIN text_entries as cte on cte.documentId = d.id
|
||||
// LEFT JOIN results as r
|
||||
// on r.queryId = dt.queryId
|
||||
// and r.queryVersion = dt.queryVersion
|
||||
// and cte.id = r.textEntryId
|
||||
// where r.value is null
|
||||
// )
|
||||
// SELECT DISTINCT queryId FROM unsyncedQueries as baseuq
|
||||
// WHERE NOT EXISTS (
|
||||
// SELECT 1 FROM unsyncedQueries as uq WHERE uq.queryId = any(baseuq.requiredIds)
|
||||
// )
|
||||
func (q *Queries) ListUnsyncedNoDepsQueriesByDocId(ctx context.Context, dollar_1 *uuid.UUID) ([]*uuid.UUID, error) {
|
||||
rows, err := q.db.Query(ctx, listUnsyncedNoDepsQueriesByDocId, dollar_1)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []*uuid.UUID{}
|
||||
for rows.Next() {
|
||||
var queryid *uuid.UUID
|
||||
if err := rows.Scan(&queryid); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, queryid)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
@@ -1,533 +0,0 @@
|
||||
package repository_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"queryorchestration/internal/database/repository"
|
||||
"queryorchestration/internal/serviceconfig"
|
||||
"queryorchestration/internal/test"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestResults(t *testing.T) {
|
||||
t.Parallel()
|
||||
if testing.Short() {
|
||||
t.SkipNow()
|
||||
}
|
||||
ctx := t.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
queryId := createQuery(t, queries, repository.QuerytypeContextFull)
|
||||
clientId := createClientWithCollector(t, queries)
|
||||
documentID, _, _ := createDocumentWithCollectorAndResult(t, queries, clientId, queryId)
|
||||
|
||||
qv := int32(1)
|
||||
res, err := queries.GetResultValueWithVersion(ctx, &repository.GetResultValueWithVersionParams{
|
||||
Queryid: &queryId,
|
||||
Queryversion: &qv,
|
||||
Documentid: &documentID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.NotNil(t, res.Value)
|
||||
assert.Equal(t, "json_value", *res.Value)
|
||||
}
|
||||
|
||||
func TestResultValues(t *testing.T) {
|
||||
t.Parallel()
|
||||
if testing.Short() {
|
||||
t.SkipNow()
|
||||
}
|
||||
ctx := t.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor))
|
||||
require.NoError(t, err)
|
||||
|
||||
clientId := "EXAMPLE"
|
||||
err = queries.CreateClient(ctx, &repository.CreateClientParams{
|
||||
Name: "example_client",
|
||||
Clientid: clientId,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
documentID, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{
|
||||
Clientid: clientId,
|
||||
Hash: "example_hash",
|
||||
BatchID: nil,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
documentTwoID, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{
|
||||
Clientid: clientId,
|
||||
Hash: "example_hash_two",
|
||||
BatchID: nil,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
contextQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull))
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = queries.AddLatestQueryVersion(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
err = queries.AddActiveQueryVersion(ctx, &repository.AddActiveQueryVersionParams{
|
||||
Queryid: jsonQueryID,
|
||||
Versionid: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
_, err = queries.AddLatestQueryVersion(ctx, contextQueryID)
|
||||
require.NoError(t, err)
|
||||
_, err = queries.AddLatestQueryVersion(ctx, contextQueryID)
|
||||
require.NoError(t, err)
|
||||
err = queries.AddActiveQueryVersion(ctx, &repository.AddActiveQueryVersionParams{
|
||||
Queryid: contextQueryID,
|
||||
Versionid: 2,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
jsonVersion := int32(1)
|
||||
err = queries.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
|
||||
Queryid: jsonQueryID,
|
||||
Requiredqueryid: contextQueryID,
|
||||
Addedversion: jsonVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
contextQuery, err := queries.GetQuery(ctx, contextQueryID)
|
||||
require.NoError(t, err)
|
||||
|
||||
qResults, err := queries.ListQueryRequirementValues(ctx, &repository.ListQueryRequirementValuesParams{
|
||||
Queryid: &jsonQueryID,
|
||||
Documentid: &documentID,
|
||||
Version: &jsonVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qResults, 0)
|
||||
qResults, err = queries.ListQueryRequirementValues(ctx, &repository.ListQueryRequirementValuesParams{
|
||||
Queryid: &jsonQueryID,
|
||||
Documentid: &documentTwoID,
|
||||
Version: &jsonVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qResults, 0)
|
||||
|
||||
version, err := queries.AddLatestCollectorVersion(ctx, clientId)
|
||||
require.NoError(t, err)
|
||||
err = queries.SetActiveCollectorVersion(ctx, &repository.SetActiveCollectorVersionParams{
|
||||
Versionid: version,
|
||||
Clientid: clientId,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
bucket := "example_bucket"
|
||||
key := "example_key"
|
||||
hash := "hash"
|
||||
cleanid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{
|
||||
Documentid: documentID,
|
||||
Bucket: &bucket,
|
||||
Key: &key,
|
||||
Hash: &hash,
|
||||
Mimetype: repository.NullCleanmimetype{
|
||||
Valid: true,
|
||||
Cleanmimetype: repository.CleanmimetypeApplicationPdf,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{
|
||||
Cleanid: cleanid,
|
||||
Version: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
textId, err := queries.AddDocumentText(ctx, &repository.AddDocumentTextParams{
|
||||
Cleanid: cleanid,
|
||||
Bucket: "hi",
|
||||
Key: "hello",
|
||||
Hash: "example",
|
||||
Part: 0,
|
||||
Createdat: pgtype.Timestamp{
|
||||
Time: time.Now().UTC(),
|
||||
Valid: true,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
|
||||
Version: 1,
|
||||
Textid: textId,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
result := repository.AddResultParams{
|
||||
Queryid: contextQueryID,
|
||||
Value: "context_value_1",
|
||||
Textentryid: textId,
|
||||
Queryversion: contextQuery.Activeversion,
|
||||
}
|
||||
|
||||
_, err = queries.AddResult(ctx, &result)
|
||||
require.NoError(t, err)
|
||||
|
||||
qResults, err = queries.ListQueryRequirementValues(ctx, &repository.ListQueryRequirementValuesParams{
|
||||
Queryid: &jsonQueryID,
|
||||
Documentid: &documentID,
|
||||
Version: &jsonVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qResults, 1)
|
||||
assert.Equal(t, contextQueryID, qResults[0].Queryid)
|
||||
assert.Equal(t, repository.QuerytypeContextFull, qResults[0].Querytype)
|
||||
assert.Equal(t, "context_value_1", *qResults[0].Value)
|
||||
assert.NotEqual(t, uuid.UUID{}, qResults[0].ID)
|
||||
qResults, err = queries.ListQueryRequirementValues(ctx, &repository.ListQueryRequirementValuesParams{
|
||||
Queryid: &jsonQueryID,
|
||||
Documentid: &documentTwoID,
|
||||
Version: &jsonVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qResults, 0)
|
||||
|
||||
_, err = queries.AddResult(ctx, &repository.AddResultParams{
|
||||
Queryid: contextQueryID,
|
||||
Value: "context_value_2",
|
||||
Textentryid: textId,
|
||||
Queryversion: contextQuery.Activeversion - 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qResults, err = queries.ListQueryRequirementValues(ctx, &repository.ListQueryRequirementValuesParams{
|
||||
Queryid: &jsonQueryID,
|
||||
Documentid: &documentID,
|
||||
Version: &jsonVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qResults, 1)
|
||||
assert.Equal(t, contextQueryID, qResults[0].Queryid)
|
||||
assert.Equal(t, repository.QuerytypeContextFull, qResults[0].Querytype)
|
||||
assert.Equal(t, "context_value_1", *qResults[0].Value)
|
||||
assert.NotEqual(t, uuid.UUID{}, qResults[0].ID)
|
||||
|
||||
_, err = queries.AddResult(ctx, &repository.AddResultParams{
|
||||
Queryid: contextQueryID,
|
||||
Value: "context_value_3",
|
||||
Textentryid: textId,
|
||||
Queryversion: contextQuery.Activeversion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qResults, err = queries.ListQueryRequirementValues(ctx, &repository.ListQueryRequirementValuesParams{
|
||||
Queryid: &jsonQueryID,
|
||||
Documentid: &documentID,
|
||||
Version: &jsonVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qResults, 1)
|
||||
assert.Equal(t, contextQueryID, qResults[0].Queryid)
|
||||
assert.Equal(t, repository.QuerytypeContextFull, qResults[0].Querytype)
|
||||
assert.Equal(t, "context_value_3", *qResults[0].Value)
|
||||
assert.NotEqual(t, uuid.UUID{}, qResults[0].ID)
|
||||
|
||||
_, err = queries.AddResult(ctx, &repository.AddResultParams{
|
||||
Queryid: jsonQueryID,
|
||||
Value: "json_value_1",
|
||||
Textentryid: textId,
|
||||
Queryversion: jsonVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qResults, err = queries.ListQueryRequirementValues(ctx, &repository.ListQueryRequirementValuesParams{
|
||||
Queryid: &jsonQueryID,
|
||||
Documentid: &documentID,
|
||||
Version: &jsonVersion,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qResults, 1)
|
||||
assert.Equal(t, contextQueryID, qResults[0].Queryid)
|
||||
assert.Equal(t, repository.QuerytypeContextFull, qResults[0].Querytype)
|
||||
assert.Equal(t, "context_value_3", *qResults[0].Value)
|
||||
assert.NotEqual(t, uuid.UUID{}, qResults[0].ID)
|
||||
}
|
||||
|
||||
func TestUnsyncedNoDepsQueries(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.SkipNow()
|
||||
}
|
||||
t.Run("single document", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := t.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
clientId := createClientWithCollector(t, queries)
|
||||
contextQueryID, jsonQueryID := createDependentQueries(t, ctx, queries)
|
||||
|
||||
err := queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
|
||||
Clientid: clientId,
|
||||
Name: "example_name",
|
||||
Queryid: jsonQueryID,
|
||||
Addedversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
documentID, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{
|
||||
Clientid: clientId,
|
||||
Hash: "example_hash",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err := queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 0)
|
||||
|
||||
bucket := "example_bucket"
|
||||
key := "example_key"
|
||||
hash := "hahs"
|
||||
cleanid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{
|
||||
Documentid: documentID,
|
||||
Bucket: &bucket,
|
||||
Key: &key,
|
||||
Hash: &hash,
|
||||
Mimetype: repository.NullCleanmimetype{
|
||||
Valid: true,
|
||||
Cleanmimetype: repository.CleanmimetypeApplicationPdf,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{
|
||||
Cleanid: cleanid,
|
||||
Version: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 0)
|
||||
|
||||
textId, err := queries.AddDocumentText(ctx, &repository.AddDocumentTextParams{
|
||||
Cleanid: cleanid,
|
||||
Bucket: "hi",
|
||||
Key: "hello",
|
||||
Part: 0,
|
||||
Createdat: pgtype.Timestamp{
|
||||
Time: time.Now().UTC(),
|
||||
Valid: true,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
|
||||
Version: 1,
|
||||
Textid: textId,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 1)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{&contextQueryID}, qs)
|
||||
|
||||
_, err = queries.AddResult(ctx, &repository.AddResultParams{
|
||||
Queryid: contextQueryID,
|
||||
Value: "context_value",
|
||||
Textentryid: textId,
|
||||
Queryversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 1)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{&jsonQueryID}, qs)
|
||||
|
||||
_, err = queries.AddResult(ctx, &repository.AddResultParams{
|
||||
Queryid: jsonQueryID,
|
||||
Value: "context_value",
|
||||
Textentryid: textId,
|
||||
Queryversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 0)
|
||||
|
||||
_, err = queries.AddLatestQueryVersion(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
err = queries.AddActiveQueryVersion(ctx, &repository.AddActiveQueryVersionParams{
|
||||
Queryid: jsonQueryID,
|
||||
Versionid: 2,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 1)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{&jsonQueryID}, qs)
|
||||
})
|
||||
|
||||
t.Run("two files", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := t.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
clientId := createClientWithCollector(t, queries)
|
||||
contextQueryID, jsonQueryID := createDependentQueries(t, ctx, queries)
|
||||
documentID, _, _ := createDocumentWithCollectorAndResults(t, queries, clientId, contextQueryID, jsonQueryID)
|
||||
|
||||
documentTwoID, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{
|
||||
Clientid: clientId,
|
||||
Hash: "example_hash_two",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
bucket := "example_bucket"
|
||||
key := "example_key"
|
||||
hash := "hahs"
|
||||
|
||||
cleantwoid, err := queries.AddDocumentClean(ctx, &repository.AddDocumentCleanParams{
|
||||
Documentid: documentTwoID,
|
||||
Bucket: &bucket,
|
||||
Key: &key,
|
||||
Hash: &hash,
|
||||
Mimetype: repository.NullCleanmimetype{
|
||||
Valid: true,
|
||||
Cleanmimetype: repository.CleanmimetypeApplicationPdf,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{
|
||||
Cleanid: cleantwoid,
|
||||
Version: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err := queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 0)
|
||||
qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentTwoID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 0)
|
||||
|
||||
textTwoId, err := queries.AddDocumentText(ctx, &repository.AddDocumentTextParams{
|
||||
Cleanid: cleantwoid,
|
||||
Bucket: "hi",
|
||||
Key: "hello",
|
||||
Part: 0,
|
||||
Createdat: pgtype.Timestamp{
|
||||
Time: time.Now().UTC(),
|
||||
Valid: true,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{
|
||||
Version: 1,
|
||||
Textid: textTwoId,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = queries.AddResult(ctx, &repository.AddResultParams{
|
||||
Queryid: contextQueryID,
|
||||
Value: "context_value",
|
||||
Textentryid: textTwoId,
|
||||
Queryversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 0)
|
||||
qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentTwoID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 1)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{&jsonQueryID}, qs)
|
||||
|
||||
_, err = queries.AddResult(ctx, &repository.AddResultParams{
|
||||
Queryid: jsonQueryID,
|
||||
Value: "context_value",
|
||||
Textentryid: textTwoId,
|
||||
Queryversion: 1,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 0)
|
||||
qs, err = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentTwoID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 0)
|
||||
})
|
||||
|
||||
t.Run("update query", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := t.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(t, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
clientId := createClientWithCollector(t, queries)
|
||||
contextQueryID, jsonQueryID := createDependentQueries(t, ctx, queries)
|
||||
documentID, _, _ := createDocumentWithCollectorAndResults(t, queries, clientId, contextQueryID, jsonQueryID)
|
||||
|
||||
_, err := queries.AddLatestQueryVersion(ctx, jsonQueryID)
|
||||
require.NoError(t, err)
|
||||
err = queries.AddActiveQueryVersion(ctx, &repository.AddActiveQueryVersionParams{
|
||||
Queryid: jsonQueryID,
|
||||
Versionid: 2,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
qs, err := queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, qs, 1)
|
||||
assert.ElementsMatch(t, []*uuid.UUID{&jsonQueryID}, qs)
|
||||
})
|
||||
}
|
||||
|
||||
func BenchmarkListUnsynced(b *testing.B) {
|
||||
ctx := b.Context()
|
||||
|
||||
cfg := &serviceconfig.BaseConfig{}
|
||||
test.CreateDB(b, cfg)
|
||||
|
||||
queries := cfg.GetDBQueries()
|
||||
|
||||
clientId := createClientWithCollector(b, queries)
|
||||
_, jsonQueryID := createDependentQueries(b, ctx, queries)
|
||||
|
||||
err := queries.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
|
||||
Clientid: clientId,
|
||||
Name: "example_name",
|
||||
Queryid: jsonQueryID,
|
||||
Addedversion: 1,
|
||||
})
|
||||
require.NoError(b, err)
|
||||
|
||||
documentID, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{
|
||||
Clientid: clientId,
|
||||
Hash: "example_hash",
|
||||
BatchID: nil,
|
||||
})
|
||||
require.NoError(b, err)
|
||||
|
||||
b.ResetTimer()
|
||||
|
||||
for b.Loop() {
|
||||
_, _ = queries.ListUnsyncedNoDepsQueriesByDocId(ctx, &documentID)
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user