Skip to content

Commit 8603720

Browse files
salamonpavelsalamonpavel
authored
#303: Get flow checkpoints refactoring (#304)
refactoring of get flow checkpoints function
1 parent 5491e5c commit 8603720

21 files changed

Lines changed: 834 additions & 68 deletions

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,3 +91,4 @@ utils/resources/*.conf
9191
/server/certs/
9292
/server/selfsigned.crt
9393
/server/selfsigned.p12
94+
/.bloop/

‎database/src/main/postgres/flows/V0.2.0.57__get_flow_checkpoints.sql‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ CREATE OR REPLACE FUNCTION flows.get_flow_checkpoints(
3333
OUT has_more BOOLEAN
3434
) RETURNS SETOF record AS
3535
$$
36-
--------------------------------------------------------------------------------------------------------------------
36+
--------------------------------------------------------------------------------------------------------------------
3737
--
3838
-- Function: flows.get_flow_checkpoints(4)
3939
-- Retrieves all checkpoints (measures and their measurement details) related to a primary flow
@@ -144,6 +144,6 @@ BEGIN
144144
LC.id_checkpoint, LC.process_start_time;
145145
END;
146146
$$
147-
LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
147+
LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
148148

149149
GRANT EXECUTE ON FUNCTION flows.get_flow_checkpoints(BIGINT, INT, BIGINT, TEXT) TO atum_owner;

‎database/src/main/postgres/flows/V0.3.0.1__get_flow_checkpoints.sql‎

Lines changed: 182 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,182 @@
1+
/*
2+
* Copyright 2021 ABSA Group Limited
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
DROP FUNCTION IF EXISTS flows.get_flow_checkpoints(
18+
i_flow_id BIGINT,
19+
i_checkpoints_limit INT,
20+
i_offset BIGINT,
21+
i_checkpoint_name TEXT
22+
);
23+
24+
CREATE OR REPLACE FUNCTION flows.get_flow_checkpoints(
25+
IN i_flow_id BIGINT,
26+
IN i_checkpoints_limit INT DEFAULT 5,
27+
IN i_offset BIGINT DEFAULT 0,
28+
IN i_checkpoint_name TEXT DEFAULT NULL,
29+
IN i_latest_first BOOLEAN DEFAULT TRUE,
30+
OUT status INTEGER,
31+
OUT status_text TEXT,
32+
OUT id_checkpoint UUID,
33+
OUT checkpoint_name TEXT,
34+
OUT checkpoint_author TEXT,
35+
OUT measured_by_atum_agent BOOLEAN,
36+
OUT measure_name TEXT,
37+
OUT measured_columns TEXT[],
38+
OUT measurement_value JSONB,
39+
OUT checkpoint_start_time TIMESTAMP WITH TIME ZONE,
40+
OUT checkpoint_end_time TIMESTAMP WITH TIME ZONE,
41+
OUT id_partitioning BIGINT,
42+
OUT partitioning JSONB,
43+
OUT partitioning_author TEXT,
44+
OUT has_more BOOLEAN
45+
) RETURNS SETOF record AS
46+
$$
47+
--------------------------------------------------------------------------------------------------------------------
48+
--
49+
-- Function: flows.get_flow_checkpoints(4)
50+
-- Retrieves all checkpoints (measures and their measurement details) related to a primary flow
51+
-- associated with the input partitioning.
52+
--
53+
-- Note: a single row returned from this function doesn't contain all data related to a single checkpoint - it only
54+
-- represents one measure associated with a checkpoint. So even if only a single checkpoint would be retrieved,
55+
-- this function can potentially return multiple rows.
56+
--
57+
-- Note: checkpoints will be retrieved in ordered fashion, by checkpoint_time and id_checkpoint
58+
--
59+
-- Parameters:
60+
-- i_partitioning_of_flow - partitioning to use for identifying the flow associate with checkpoints
61+
-- that will be retrieved
62+
-- i_checkpoints_limit - (optional) maximum number of checkpoint to return, returns all of them if NULL
63+
-- i_offset - (optional) offset for checkpoints pagination
64+
-- i_checkpoint_name - (optional) if specified, returns data related to particular checkpoint's name
65+
-- i_latest_first - (optional) if true, checkpoints are ordered by process_start_time in descending order
66+
--
67+
-- Note: i_checkpoint_limit and i_offset are used for pagination purposes;
68+
-- checkpoints are ordered by process_start_time in descending order
69+
-- and then by id_checkpoint in ascending order
70+
--
71+
-- Returns:
72+
-- status - Status code
73+
-- status_text - Status text
74+
-- id_checkpoint - ID of retrieved checkpoint
75+
-- checkpoint_name - Name of the retrieved checkpoint
76+
-- checkpoint_author - Author of the checkpoint
77+
-- measured_by_atum_agent - Flag indicating whether the checkpoint was measured by Atum Agent
78+
-- (if false, data supplied manually)
79+
-- measure_name - measure name associated with a given checkpoint
80+
-- measured_columns - measure columns associated with a given checkpoint
81+
-- measurement_value - measurement details associated with a given checkpoint
82+
-- checkpoint_start_time - Time of the checkpoint
83+
-- checkpoint_end_time - End time of the checkpoint computation
84+
-- id_partitioning - ID of the partitioning
85+
-- partitioning - Partitioning value
86+
-- partitioning_author - Author of the partitioning
87+
-- has_more - flag indicating whether there are more checkpoints available, always `false` if `i_limit` is NULL
88+
--
89+
-- Status codes:
90+
-- 11 - OK
91+
-- 42 - Flow not found
92+
---------------------------------------------------------------------------------------------------
93+
DECLARE
94+
_has_more BOOLEAN;
95+
_latest_first BOOLEAN := coalesce(i_latest_first, TRUE);
96+
BEGIN
97+
-- Check if the flow exists by querying the partitioning_to_flow table.
98+
-- Rationale:
99+
-- This table is preferred over the flows table because:
100+
-- 1. Every flow has at least one record in partitioning_to_flow.
101+
-- 2. This table is used in subsequent queries, providing a caching advantage.
102+
-- 3. Improves performance by reducing the need to query the flows table directly.
103+
PERFORM 1 FROM flows.partitioning_to_flow WHERE fk_flow = i_flow_id;
104+
IF NOT FOUND THEN
105+
status := 42;
106+
status_text := 'Flow not found';
107+
RETURN NEXT;
108+
RETURN;
109+
END IF;
110+
111+
-- Determine if there are more checkpoints than the limit
112+
IF i_checkpoints_limit IS NOT NULL THEN
113+
SELECT count(*) > i_checkpoints_limit
114+
FROM runs.checkpoints C
115+
JOIN flows.partitioning_to_flow PF ON C.fk_partitioning = PF.fk_partitioning
116+
WHERE PF.fk_flow = i_flow_id
117+
AND (i_checkpoint_name IS NULL OR C.checkpoint_name = i_checkpoint_name)
118+
LIMIT i_checkpoints_limit + 1 OFFSET i_offset
119+
INTO _has_more;
120+
ELSE
121+
_has_more := false;
122+
END IF;
123+
124+
-- Retrieve the checkpoints and their associated measurements
125+
RETURN QUERY
126+
WITH limited_checkpoints AS (
127+
SELECT C.id_checkpoint,
128+
C.fk_partitioning,
129+
C.checkpoint_name,
130+
C.created_by,
131+
C.measured_by_atum_agent,
132+
C.process_start_time,
133+
C.process_end_time
134+
FROM runs.checkpoints C
135+
JOIN flows.partitioning_to_flow PF ON C.fk_partitioning = PF.fk_partitioning
136+
WHERE PF.fk_flow = i_flow_id
137+
AND (i_checkpoint_name IS NULL OR C.checkpoint_name = i_checkpoint_name)
138+
ORDER BY
139+
CASE
140+
WHEN _latest_first THEN C.process_start_time
141+
END DESC,
142+
CASE
143+
WHEN NOT _latest_first THEN C.process_start_time
144+
END ASC
145+
LIMIT i_checkpoints_limit OFFSET i_offset
146+
)
147+
SELECT
148+
11 AS status,
149+
'OK' AS status_text,
150+
LC.id_checkpoint,
151+
LC.checkpoint_name,
152+
LC.created_by AS author,
153+
LC.measured_by_atum_agent,
154+
MD.measure_name,
155+
MD.measured_columns,
156+
M.measurement_value,
157+
LC.process_start_time AS checkpoint_start_time,
158+
LC.process_end_time AS checkpoint_end_time,
159+
LC.fk_partitioning AS id_partitioning,
160+
P.partitioning AS partitioning,
161+
P.created_by AS partitioning_author,
162+
_has_more AS has_more
163+
FROM
164+
limited_checkpoints LC
165+
INNER JOIN
166+
runs.measurements M ON LC.id_checkpoint = M.fk_checkpoint
167+
INNER JOIN
168+
runs.measure_definitions MD ON M.fk_measure_definition = MD.id_measure_definition
169+
INNER JOIN
170+
runs.partitionings P ON LC.fk_partitioning = P.id_partitioning
171+
ORDER BY
172+
CASE
173+
WHEN _latest_first THEN LC.process_start_time
174+
END DESC,
175+
CASE
176+
WHEN NOT _latest_first THEN LC.process_start_time
177+
END ASC;
178+
END;
179+
$$
180+
LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
181+
182+
GRANT EXECUTE ON FUNCTION flows.get_flow_checkpoints(BIGINT, INT, BIGINT, TEXT, BOOLEAN) TO atum_owner;

0 commit comments

Comments
 (0)