#478 (2): checkpoint time window and latest-first on flows (audit consumption pattern)

README.md

@@ -108,12 +108,13 @@ drop support for Spark 2.4. If you need the agent to work with Spark 2.4 follow
 * Build these two modules and use them in your project
 
 ### Reader `reader/`
-**NB!**  
-_This module is not yet implemented to an operational abilities and therefore not yet released._
+This module is a Scala client for applications reading metrics stored by the _Atum Service_. `FlowReader` and
+`PartitioningReader` resolve partitionings and retrieve checkpoint pages through the server's REST API.
+Applications control pagination; the Reader does not discover datasets from partial business keys or reconstruct
+the original parent-child lineage graph.
 
-This module is intended to be used whenever an application wants to read the metrics stored by the _Atum Service_. It
-offers classes and methods to read the metrics from the database shielding away the complexity of accessing the _Atum Server_
-directly.
+Reader features require a compatible server version. In particular, older servers can ignore checkpoint time-window
+parameters; using a newer Reader alone does not enable server-side filtering.
 
 ### Server `server/`
 An API under construction that communicates with the Agent and with the persistent storage. It also provides measure 

agent/README.md

@@ -25,7 +25,7 @@ Use `AtumPartitions` to get an `AtumContext` from the service using the `AtumAge
 import za.co.absa.atum.agent.AtumAgent
 import za.co.absa.atum.model.types.basic.AtumPartitions
 
-val atumPartitions = AtumPartitions("country" -> "za", "dataset" -> "customers")
+val atumPartitions = AtumPartitions(List("country" -> "za", "dataset" -> "customers"))
 val atumContext1 = AtumAgent.getOrCreateAtumContext(atumPartitions)
 ```
 
@@ -45,19 +45,26 @@ val config = ConfigFactory.parseString(
 )
 
 val customAgent = AtumAgent.fromConfig(config)
-val atumPartitions = AtumPartitions("country" -> "za", "dataset" -> "customers")
+val atumPartitions = AtumPartitions(List("country" -> "za", "dataset" -> "customers"))
 val atumContext = customAgent.getOrCreateAtumContext(atumPartitions)
 ```
 
 #### AtumPartitions
 A list of key values that maintains the order of arrival of the items, the `AtumService`
 is able to deliver the correct `AtumContext` according to the `AtumPartitions` we give it.
 ```scala
-    val atumPartitions = AtumPartitions().withPartitions(ListMap("name" -> "partition-name", "country" -> "SA", "gender" -> "female" ))
+val atumPartitions = AtumPartitions(List(
+  "name" -> "partition-name",
+  "country" -> "SA",
+  "gender" -> "female"
+))
 
-    val subPartition = atumPartitions.addPartition("otherKey", "otherValue")
+val withOtherKey = atumPartitions ++ AtumPartitions("otherKey" -> "otherValue")
 ```
 
+Constructing these keys does not register a parent-child relationship. Use
+`AtumContext.subPartitionContext` to create a linked child context.
+
 Control measures can also be overwritten, added or removed.
 
 ```scala

database/src/main/postgres/flows/V0.8.0.5__get_flow_partitionings.sql

@@ -0,0 +1,98 @@
+/*
+ * 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.
+ */
+
+CREATE OR REPLACE FUNCTION flows.get_flow_partitionings(
+    IN i_flow_id               BIGINT,
+    IN i_limit                 INT DEFAULT 5,
+    IN i_offset                BIGINT DEFAULT 0,
+    OUT status                 INTEGER,
+    OUT status_text            TEXT,
+    OUT id                     BIGINT,
+    OUT partitioning           JSONB,
+    OUT author                 TEXT,
+    OUT has_more               BOOLEAN
+) RETURNS SETOF record AS
+-------------------------------------------------------------------------------
+--
+-- Function: flows.get_flow_partitionings(3)
+--      Retrieves all partitionings associated with the input flow.
+--
+-- Note: partitionings are ordered by id_partitioning, then created_at descending.
+--
+-- Parameters:
+--      i_flow_id               - flow id to use for identifying the partitionings that will be retrieved
+--      i_limit                 - (optional) maximum number of partitionings to return, default is 5
+--      i_offset                - (optional) offset to use for pagination, default is 0
+--
+-- Returns:
+--      status                  - Status code
+--      status_text             - Status text
+--      id                      - ID of retrieved partitioning
+--      partitioning            - Partitioning value
+--      author                  - Author of the partitioning
+--      has_more                - Flag indicating if there are more partitionings available
+--
+-- Status codes:
+--      11 - OK
+--      41 - Flow not found
+--
+-------------------------------------------------------------------------------
+$$
+DECLARE
+    _has_more BOOLEAN;
+BEGIN
+    PERFORM 1 FROM flows.flows WHERE id_flow = i_flow_id;
+    IF NOT FOUND THEN
+        status := 41;
+        status_text := 'Flow not found';
+        RETURN NEXT;
+        RETURN;
+    END IF;
+
+    IF i_limit IS NOT NULL THEN
+        SELECT count(*) > i_limit
+        FROM (SELECT 1
+              FROM flows.partitioning_to_flow PTF
+              WHERE PTF.fk_flow = i_flow_id
+              LIMIT i_limit + 1 OFFSET i_offset) s
+        INTO _has_more;
+    ELSE
+        _has_more := false;
+    END IF;
+
+
+    RETURN QUERY
+        SELECT
+            11 AS status,
+            'OK' AS status_text,
+            P.id_partitioning,
+            P.partitioning,
+            P.created_by,
+            _has_more
+        FROM
+            runs.partitionings P INNER JOIN
+            flows.partitioning_to_flow PF ON  PF.fk_partitioning = P.id_partitioning
+        WHERE
+            PF.fk_flow = i_flow_id
+        ORDER BY
+            P.id_partitioning,
+            P.created_at DESC
+        LIMIT i_limit OFFSET i_offset;
+END;
+$$
+LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
+
+GRANT EXECUTE ON FUNCTION flows.get_flow_partitionings(BIGINT, INT, BIGINT) TO atum_owner;

database/src/main/postgres/runs/V0.8.0.4__checkpoints_partitioning_start_time_index.ddl

@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+
+-- Supports the process start time window of runs.get_partitioning_checkpoints and flows.get_flow_checkpoints:
+-- checkpoints of each partitioning are range-scanned by process_start_time.
+--
+-- Built CONCURRENTLY so writes to runs.checkpoints are not blocked. This file must contain only this statement:
+-- Flyway then runs it outside of a transaction, which CONCURRENTLY requires.
+-- If the build fails, it leaves an INVALID index behind; drop it (DROP INDEX CONCURRENTLY runs.checkpoints_idx2)
+-- and re-run the migration.
+CREATE INDEX CONCURRENTLY checkpoints_idx2 ON runs.checkpoints (fk_partitioning, process_start_time);

database/src/test/scala/za/co/absa/atum/database/flows/GetFlowPartitioningsIntegrationTests.scala

@@ -174,6 +174,72 @@ class GetFlowPartitioningsIntegrationTests extends DBTestSuite {
       }
   }
 
+  Seq(
+    ("first page", Some(2), 0L, Seq(0, 1), true),
+    ("middle page", Some(2), 2L, Seq(2, 3), true),
+    ("last partial page", Some(2), 4L, Seq(4), false),
+    ("exact final full page", Some(2), 3L, Seq(3, 4), false),
+    ("offset at the end", Some(2), 5L, Seq.empty[Int], false),
+    ("offset beyond the end", Some(2), 6L, Seq.empty[Int], false),
+    ("unlimited first page", None, 0L, Seq(0, 1, 2, 3, 4), false),
+    ("unlimited page with an offset", None, 2L, Seq(2, 3, 4), false)
+  ).foreach { case (description, limit, offset, expectedIndices, hasMore) =>
+    test(s"Returns ordered partitionings and has_more for $description") {
+      val partitionings = (0 until 5).map { index =>
+        val partitioning = JsonBString(
+          s"""{"version":1,"keys":["page"],"keysToValuesMap":{"page":"$index"}}"""
+        )
+        table(partitioningsTable).insert(add("partitioning", partitioning).add("created_by", "Pagination author"))
+        val id: Long = table(partitioningsTable)
+          .fieldValue("partitioning", partitioning, "id_partitioning").get.get
+        (id, partitioning)
+      }.sortBy(_._1)
+
+      val flowId = function(createFlowFn)
+        .setParam("i_fk_partitioning", partitionings.last._1)
+        .setParam("i_by_user", "Pagination author")
+        .execute { queryResult =>
+          val row = queryResult.next()
+          assert(row.getInt("status").contains(11))
+          row.getLong("id_flow").get
+        }
+
+      partitionings.init.reverse.foreach { case (id, _) =>
+        function(addToParentFlowsFn)
+          .setParam("i_fk_parent_partitioning", partitionings.last._1)
+          .setParam("i_fk_partitioning", id)
+          .setParam("i_by_user", "Pagination author")
+          .execute { queryResult =>
+            assert(queryResult.next().getInt("status").contains(11))
+          }
+      }
+
+      val pageQuery = function(getFlowPartitioningsFn)
+        .setParam("i_flow_id", flowId)
+        .setParam("i_offset", offset)
+      val limitedQuery = limit match {
+        case Some(value) => pageQuery.setParam("i_limit", value)
+        case None => pageQuery.setParamNull("i_limit")
+      }
+
+      limitedQuery.execute { queryResult =>
+        expectedIndices.foreach { index =>
+          assert(queryResult.hasNext)
+          val row = queryResult.next()
+          val (expectedId, expectedPartitioning) = partitionings(index)
+          assert(row.getInt("status").contains(11))
+          assert(row.getString("status_text").contains("OK"))
+          assert(row.getLong("id").contains(expectedId))
+          assert(parseJsonBStringOrThrow(row.getJsonB("partitioning").get) ==
+            parseJsonBStringOrThrow(expectedPartitioning))
+          assert(row.getString("author").contains("Pagination author"))
+          assert(row.getBoolean("has_more").contains(hasMore))
+        }
+        assert(!queryResult.hasNext)
+      }
+    }
+  }
+
   test("Fails for non-existent flow"){
     function(getFlowPartitioningsFn)
       .setParam("i_flow_id", 999999)

reader/src/main/scala/za/co/absa/atum/reader/FlowReader.scala

@@ -45,12 +45,12 @@ case class FlowReader[F[_]](mainFlowPartitioning: AtumPartitions)(implicit
 
   /**
    *  Function to retrieve a page of checkpoints belonging to the flow, optionally filtered.
-   *  The checkpoints are ordered by their process start time, latest first.
+   *  The checkpoints are ordered by their process start time, latest first unless the filter says otherwise.
    *
    *  @param pageSize          - the size of the page (record count) to be returned
    *  @param offset            - offset of the page (starting position)
    *  @param includeProperties - whether to include checkpoint properties in the response
-   *  @param filter            - the checkpoints to return (by name and/or properties)
+   *  @param filter            - the checkpoints to return (name, properties, process start time window) and their order
    *  @return                  - a page of checkpoints
    */
   def getCheckpointsPage(

reader/src/main/scala/za/co/absa/atum/reader/PartitioningReader.scala

@@ -43,12 +43,12 @@ case class PartitioningReader[F[_]](partitioning: AtumPartitions)(implicit
 
   /**
    *  Function to retrieve a page of checkpoints belonging to the partitioning, optionally filtered.
-   *  The checkpoints are ordered by their process start time, latest first.
+   *  The checkpoints are ordered by their process start time, latest first unless the filter says otherwise.
    *
    *  @param pageSize          - the size of the page (record count) to be returned
    *  @param offset            - offset of the page (starting position)
    *  @param includeProperties - whether to include checkpoint properties in the response
-   *  @param filter            - the checkpoints to return (by name and/or properties)
+   *  @param filter            - the checkpoints to return (name, properties, process start time window) and their order
    *  @return                  - a page of checkpoints
    */
   def getCheckpointsPage(

reader/src/main/scala/za/co/absa/atum/reader/requests/CheckpointFilter.scala

@@ -18,27 +18,37 @@ package za.co.absa.atum.reader.requests
 
 import za.co.absa.atum.model.utils.JsonSyntaxExtensions._
 
+import java.time.ZonedDateTime
+
 /**
  *  Filter of a checkpoints query. All its parts are optional and a checkpoint has to satisfy all the given ones.
  *
- *  Example - checkpoints of the given name, of any of the two executions:
+ *  Example - checkpoints of two executions processed in the last two months, latest first:
  *  {{{
  *    CheckpointFilter(
- *      name = Some("Data written"),
- *      properties = Map("executionID" -> Set("a", "b"))
+ *      properties = Map("executionID" -> Set("a", "b")),
+ *      from = Some(ZonedDateTime.now().minusMonths(2))
  *    )
  *  }}}
  *
- *  Note: properties with more than one value require an Atum server supporting them; older servers reject them.
+ *  Note: the time window and properties with more than one value require an Atum server supporting them; older servers
+ *  ignore the time window and reject multi-value properties.
  *
- *  @param name       - only checkpoints of this name
- *  @param properties - only checkpoints that, for every property name given, have that property with one of the given
- *                      values, e.g. `Map("executionID" -> Set("a", "b"))` means `executionID IN (a, b)`; every property
- *                      needs at least one value, otherwise an `IllegalArgumentException` is thrown
+ *  @param name        - only checkpoints of this name
+ *  @param properties  - only checkpoints that, for every property name given, have that property with one of the given
+ *                       values, e.g. `Map("executionID" -> Set("a", "b"))` means `executionID IN (a, b)`; every property
+ *                       needs at least one value, otherwise an `IllegalArgumentException` is thrown
+ *  @param from        - only checkpoints with process start time at or after this time (inclusive)
+ *  @param to          - only checkpoints with process start time before this time (exclusive)
+ *  @param latestFirst - order of the checkpoints by their process start time: latest first if `true`, earliest first if
+ *                       `false`, server default (latest first) if not specified
  */
 case class CheckpointFilter(
   name: Option[String] = None,
-  properties: Map[String, Set[String]] = Map.empty
+  properties: Map[String, Set[String]] = Map.empty,
+  from: Option[ZonedDateTime] = None,
+  to: Option[ZonedDateTime] = None,
+  latestFirst: Option[Boolean] = None
 ) {
 
   require(
@@ -47,7 +57,11 @@ case class CheckpointFilter(
   )
 
   private[reader] def toQueryParams: Map[String, String] = {
-    name.map(QueryParamNames.CheckpointName -> _).toMap ++ propertiesParam
+    name.map(QueryParamNames.CheckpointName -> _).toMap ++
+      propertiesParam ++
+      from.map(QueryParamNames.From -> _.toInstant.toString) ++
+      to.map(QueryParamNames.To -> _.toInstant.toString) ++
+      latestFirst.map(QueryParamNames.LatestFirst -> _.toString)
   }
 
   private def propertiesParam: Option[(String, String)] = {

reader/src/main/scala/za/co/absa/atum/reader/requests/QueryParamNames.scala

@@ -22,4 +22,7 @@ object QueryParamNames {
   final val CheckpointName = "checkpoint-name"
   final val CheckpointProperties = "checkpoint-properties"
   final val IncludeProperties = "include-properties"
+  final val LatestFirst = "latest-first"
+  final val From = "from"
+  final val To = "to"
 }

reader/src/test/scala/za/co/absa/atum/reader/FlowReaderUnitTests.scala

@@ -307,7 +307,7 @@ class FlowReaderUnitTests extends AnyFunSuiteLike {
     assert(result.isRight)
   }
 
-  test("The flow checkpoints are queried with name and multi-value properties in a single request") {
+  test("The flow checkpoints are queried with name, multi-value properties and time window in a single request") {
     implicit val server: SttpBackendStub[Identity, capabilities.WebSockets] = SttpBackendStub.synchronous
       .whenRequestMatchesPartial {
         case r if r.uri.path.endsWith(List("partitionings")) =>
@@ -320,12 +320,18 @@ class FlowReaderUnitTests extends AnyFunSuiteLike {
           assert(r.uri.querySegments.contains(KeyValue("include-properties", "true")))
           assert(r.uri.querySegments.contains(KeyValue("checkpoint-name", "Test checkpoints 1")))
           assert(r.uri.querySegments.contains(KeyValue("checkpoint-properties", multiValuePropertiesEncoded)))
+          assert(r.uri.querySegments.contains(KeyValue("from", "2026-06-01T00:00:00Z")))
+          assert(r.uri.querySegments.contains(KeyValue("to", "2026-08-01T00:00:00Z")))
+          assert(r.uri.querySegments.contains(KeyValue("latest-first", "false")))
           Response.ok(checkpointsResponse)
       }
 
     val filter = CheckpointFilter(
       name = Some("Test checkpoints 1"),
-      properties = Map("executionID" -> Set("id2", "id1"))
+      properties = Map("executionID" -> Set("id2", "id1")),
+      from = Some(ZonedDateTime.parse("2026-06-01T02:00:00+02:00")),
+      to = Some(ZonedDateTime.parse("2026-08-01T00:00:00Z")),
+      latestFirst = Some(false)
     )
     val result = FlowReader(AtumPartitions(List("a" -> "b", "c" -> "d")))
       .getCheckpointsPage(pageSize = 5, offset = 10, includeProperties = true, filter = filter)