lsulak · Oct 16, 2026 · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026
@@ -0,0 +1,221 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- The previous overload is dropped so that the new signature (JSONB multi-value checkpoint properties filter and
+-- the process start time window) is unambiguous when the function is called with named or defaulted arguments.
+DROP FUNCTION IF EXISTS flows.get_flow_checkpoints(BIGINT, INT, BIGINT, TEXT, HSTORE, BOOLEAN);
+
+CREATE OR REPLACE FUNCTION flows.get_flow_checkpoints(
+    IN i_flow_id BIGINT,
+    IN i_checkpoints_limit INT DEFAULT NULL,
+    IN i_offset BIGINT DEFAULT 0,
+    IN i_checkpoint_name TEXT DEFAULT NULL,
+    IN i_checkpoint_properties JSONB DEFAULT NULL,
+    IN i_latest_first BOOLEAN DEFAULT TRUE,
+    IN i_from_time TIMESTAMP WITH TIME ZONE DEFAULT NULL,
+    IN i_to_time TIMESTAMP WITH TIME ZONE DEFAULT NULL,
+    OUT status INTEGER,
+    OUT status_text TEXT,
+    OUT id_checkpoint UUID,
+    OUT checkpoint_name TEXT,
+    OUT checkpoint_author TEXT,
+    OUT measured_by_atum_agent BOOLEAN,
+    OUT measure_name TEXT,
+    OUT measured_columns TEXT[],
+    OUT measurement_value JSONB,
+    OUT checkpoint_start_time TIMESTAMP WITH TIME ZONE,
+    OUT checkpoint_end_time TIMESTAMP WITH TIME ZONE,
+    OUT id_partitioning BIGINT,
+    OUT partitioning JSONB,
+    OUT partitioning_author TEXT,
+    OUT has_more BOOLEAN
+) RETURNS SETOF record AS
+--------------------------------------------------------------------------------------------------------------------
+--
+-- Function: flows.get_flow_checkpoints(8)
+--      Retrieves all checkpoints (measures and their measurement details) related to an input flow (and checkpoint
+--      name, checkpoint properties and/or process start time window, if specified).
+--
+-- Note: a single row returned from this function doesn't contain all data related to a single checkpoint - it only
+--     represents one measure associated with a checkpoint. So even if only a single checkpoint would be retrieved,
+--     this function can potentially return multiple rows.
+--
+-- Parameters:
+--      i_flow_id               - ID of the flow for which checkpoints are to be retrieved
+--      i_checkpoints_limit     - (optional) maximum number of checkpoints to return, returns all of them if NULL
+--      i_offset                - (optional) offset of the first checkpoint to return
+--      i_checkpoint_name       - (optional) if specified, returns data related to particular checkpoint's name
+--      i_checkpoint_properties - (optional) JSON object mapping a property name to an array of accepted values,
+--                                  e.g. {"executionID": ["a", "b"]}; returns only checkpoints that, for every given
+--                                  property name, have that property with one of the accepted values
+--      i_latest_first          - (optional) if true (default), checkpoints are ordered by process_start_time
+--                                  in descending order (latest first); if false, in ascending order
+--      i_from_time             - (optional) if specified, returns only checkpoints with process_start_time >= i_from_time
+--      i_to_time               - (optional) if specified, returns only checkpoints with process_start_time < i_to_time
+--
+-- Note: i_checkpoints_limit and i_offset are used for pagination purposes;
+--       checkpoints are ordered by process_start_time (descending by default, see i_latest_first)
+--       and then by id_checkpoint in ascending order as a tie-breaker
+--
+-- Returns:
+--      status                 - Status code
+--      status_text            - Status text
+--      id_checkpoint          - ID of retrieved checkpoint
+--      checkpoint_name        - Name of the retrieved checkpoint
+--      checkpoint_author      - Author of the checkpoint
+--      measured_by_atum_agent - Flag indicating whether the checkpoint was measured by Atum Agent
+--                               (if false, data supplied manually)
+--      measure_name           - measure name associated with a given checkpoint
+--      measured_columns       - measure columns associated with a given checkpoint
+--      measurement_value      - measurement details associated with a given checkpoint
+--      checkpoint_start_time  - Time of the checkpoint
+--      checkpoint_end_time    - End time of the checkpoint computation
+--      id_partitioning        - ID of the partitioning
+--      partitioning           - Partitioning value
+--      partitioning_author    - Author of the partitioning
+--      has_more               - Flag indicating whether there are more checkpoints available, always `false` if `i_limit` is NULL
+--
+-- Status codes:
+--      11                     - OK
+--      42                     - Flow not found
+---------------------------------------------------------------------------------------------------
+$$
+DECLARE
+    _has_more     BOOLEAN;
+    _latest_first BOOLEAN := coalesce(i_latest_first, TRUE);
+BEGIN
+    -- Check if the flow exists by querying the partitioning_to_flow table.
+    -- Rationale:
+    -- This table is preferred over the flows table because:
+    -- 1. Every flow has at least one record in partitioning_to_flow.
+    -- 2. This table is used in subsequent queries, providing a caching advantage.
+    -- 3. Improves performance by reducing the need to query the flows table directly.
+    PERFORM 1 FROM flows.partitioning_to_flow WHERE fk_flow = i_flow_id;
+    IF NOT FOUND THEN
+        status := 42;
+        status_text := 'Flow not found';
+        RETURN NEXT;
+        RETURN;
+    END IF;
+
+    -- Determine if there are more checkpoints than the limit
+    IF i_checkpoints_limit IS NOT NULL THEN
+        SELECT count(*) > i_checkpoints_limit
+        FROM (SELECT 1
+              FROM runs.checkpoints C
+                       JOIN flows.partitioning_to_flow PF ON C.fk_partitioning = PF.fk_partitioning
+              WHERE PF.fk_flow = i_flow_id
+                AND (i_checkpoint_name IS NULL OR C.checkpoint_name = i_checkpoint_name)
+                AND (i_from_time IS NULL OR C.process_start_time >= i_from_time)
+                AND (i_to_time IS NULL OR C.process_start_time < i_to_time)
+                AND (
+                    i_checkpoint_properties IS NULL
+                    OR NOT EXISTS (
+                        SELECT 1
+                        FROM jsonb_each(i_checkpoint_properties) AS flt(k, v)
+                        WHERE NOT EXISTS (
+                            SELECT 1
+                            FROM runs.checkpoint_properties CP
+                            WHERE CP.fk_checkpoint = C.id_checkpoint
+                              AND CP.property_name = flt.k
+                              AND (v ? CP.property_value)
+                        )
+                    )
+                )
+              ORDER BY CASE
+                          WHEN _latest_first THEN C.process_start_time
+                          END DESC,
+                      CASE
+                          WHEN NOT _latest_first THEN C.process_start_time
+                          END ASC,
+                      C.id_checkpoint ASC
+              LIMIT i_checkpoints_limit + 1 OFFSET i_offset) s
+        INTO _has_more;
+    ELSE
+        _has_more := false;
+    END IF;
+
+    -- Retrieve the checkpoints and their associated measurements
+    RETURN QUERY
+        WITH limited_checkpoints AS (SELECT C.id_checkpoint,
+                                            C.fk_partitioning,
+                                            C.checkpoint_name,
+                                            C.created_by,
+                                            C.measured_by_atum_agent,
+                                            C.process_start_time,
+                                            C.process_end_time
+                                     FROM runs.checkpoints C
+                                              JOIN flows.partitioning_to_flow PF ON C.fk_partitioning = PF.fk_partitioning
+                                     WHERE PF.fk_flow = i_flow_id
+                                       AND (i_checkpoint_name IS NULL OR C.checkpoint_name = i_checkpoint_name)
+                                       AND (i_from_time IS NULL OR C.process_start_time >= i_from_time)
+                                       AND (i_to_time IS NULL OR C.process_start_time < i_to_time)
+                                       AND (
+                                           i_checkpoint_properties IS NULL
+                                           OR NOT EXISTS (
+                                               SELECT 1
+                                               FROM jsonb_each(i_checkpoint_properties) AS flt(k, v)
+                                               WHERE NOT EXISTS (
+                                                   SELECT 1
+                                                   FROM runs.checkpoint_properties CP
+                                                   WHERE CP.fk_checkpoint = C.id_checkpoint
+                                                     AND CP.property_name = flt.k
+                                                     AND (v ? CP.property_value)
+                                               )
+                                           )
+                                       )
+                                     ORDER BY CASE
+                                                  WHEN _latest_first THEN C.process_start_time
+                                                  END DESC,
+                                              CASE
+                                                  WHEN NOT _latest_first THEN C.process_start_time
+                                                  END ASC,
+                                              C.id_checkpoint ASC
+                                     LIMIT i_checkpoints_limit OFFSET i_offset)
+        SELECT 11                    AS status,
+               'OK'                  AS status_text,
+               LC.id_checkpoint,
+               LC.checkpoint_name,
+               LC.created_by         AS checkpoint_author,
+               LC.measured_by_atum_agent,
+               MD.measure_name,
+               MD.measured_columns,
+               M.measurement_value,
+               LC.process_start_time AS checkpoint_start_time,
+               LC.process_end_time   AS checkpoint_end_time,
+               LC.fk_partitioning    AS id_partitioning,
+               P.partitioning        AS partitioning,
+               P.created_by          AS partitioning_author,
+               _has_more             AS has_more
+        FROM limited_checkpoints LC
+                 INNER JOIN
+             runs.measurements M ON LC.id_checkpoint = M.fk_checkpoint
+                 INNER JOIN
+             runs.measure_definitions MD ON M.fk_measure_definition = MD.id_measure_definition
+                 INNER JOIN
+             runs.partitionings P ON LC.fk_partitioning = P.id_partitioning
+        ORDER BY CASE
+                     WHEN _latest_first THEN LC.process_start_time
+                     END DESC,
+                 CASE
+                     WHEN NOT _latest_first THEN LC.process_start_time
+                     END ASC,
+                 LC.id_checkpoint ASC;
+END;
+$$ LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
+
+ALTER FUNCTION flows.get_flow_checkpoints(BIGINT, INT, BIGINT, TEXT, JSONB, BOOLEAN, TIMESTAMP WITH TIME ZONE, TIMESTAMP WITH TIME ZONE) OWNER TO atum_owner;
+GRANT EXECUTE ON FUNCTION flows.get_flow_checkpoints(BIGINT, INT, BIGINT, TEXT, JSONB, BOOLEAN, TIMESTAMP WITH TIME ZONE, TIMESTAMP WITH TIME ZONE) TO atum_owner;