diff --git a/README.md b/README.md index 8e9880fb2..228a57284 100644 --- a/README.md +++ b/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 diff --git a/agent/README.md b/agent/README.md index 70445690c..a85705f06 100644 --- a/agent/README.md +++ b/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,7 +45,7 @@ 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) ``` @@ -53,11 +53,18 @@ val atumContext = customAgent.getOrCreateAtumContext(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 diff --git a/database/src/main/postgres/flows/V0.8.0.5__get_flow_partitionings.sql b/database/src/main/postgres/flows/V0.8.0.5__get_flow_partitionings.sql new file mode 100644 index 000000000..fd9e5dd15 --- /dev/null +++ b/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; diff --git a/database/src/main/postgres/runs/V0.8.0.4__checkpoints_partitioning_start_time_index.ddl b/database/src/main/postgres/runs/V0.8.0.4__checkpoints_partitioning_start_time_index.ddl new file mode 100644 index 000000000..9e7d84004 --- /dev/null +++ b/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); diff --git a/database/src/test/scala/za/co/absa/atum/database/flows/GetFlowPartitioningsIntegrationTests.scala b/database/src/test/scala/za/co/absa/atum/database/flows/GetFlowPartitioningsIntegrationTests.scala index 0a9745ce8..c8053fc6f 100644 --- a/database/src/test/scala/za/co/absa/atum/database/flows/GetFlowPartitioningsIntegrationTests.scala +++ b/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) diff --git a/reader/src/main/scala/za/co/absa/atum/reader/FlowReader.scala b/reader/src/main/scala/za/co/absa/atum/reader/FlowReader.scala index 4e41985ec..57f104d95 100644 --- a/reader/src/main/scala/za/co/absa/atum/reader/FlowReader.scala +++ b/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( diff --git a/reader/src/main/scala/za/co/absa/atum/reader/PartitioningReader.scala b/reader/src/main/scala/za/co/absa/atum/reader/PartitioningReader.scala index 0346788cc..5a38732f1 100644 --- a/reader/src/main/scala/za/co/absa/atum/reader/PartitioningReader.scala +++ b/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( diff --git a/reader/src/main/scala/za/co/absa/atum/reader/requests/CheckpointFilter.scala b/reader/src/main/scala/za/co/absa/atum/reader/requests/CheckpointFilter.scala index a5a497a2d..4f2c13e23 100644 --- a/reader/src/main/scala/za/co/absa/atum/reader/requests/CheckpointFilter.scala +++ b/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)] = { diff --git a/reader/src/main/scala/za/co/absa/atum/reader/requests/QueryParamNames.scala b/reader/src/main/scala/za/co/absa/atum/reader/requests/QueryParamNames.scala index f1bf111f9..e17baddcd 100644 --- a/reader/src/main/scala/za/co/absa/atum/reader/requests/QueryParamNames.scala +++ b/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" } diff --git a/reader/src/test/scala/za/co/absa/atum/reader/FlowReaderUnitTests.scala b/reader/src/test/scala/za/co/absa/atum/reader/FlowReaderUnitTests.scala index dc918b00d..e881c25c4 100644 --- a/reader/src/test/scala/za/co/absa/atum/reader/FlowReaderUnitTests.scala +++ b/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) diff --git a/reader/src/test/scala/za/co/absa/atum/reader/PartitioningReaderUnitTests.scala b/reader/src/test/scala/za/co/absa/atum/reader/PartitioningReaderUnitTests.scala index 2226e4c3e..9b7733d7f 100644 --- a/reader/src/test/scala/za/co/absa/atum/reader/PartitioningReaderUnitTests.scala +++ b/reader/src/test/scala/za/co/absa/atum/reader/PartitioningReaderUnitTests.scala @@ -277,7 +277,7 @@ class PartitioningReaderUnitTests extends AnyFunSuiteLike { assert(result.isRight) } - test("The partitioning checkpoints are queried with name and multi-value properties in a single request") { + test("The partitioning 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(V2Paths.Partitionings)) => @@ -288,12 +288,16 @@ class PartitioningReaderUnitTests extends AnyFunSuiteLike { assert(r.uri.querySegments.contains(KeyValue("checkpoint-name", "Test checkpoints 1"))) // base64url-encoded {"executionID":["id1","id2"]} assert(r.uri.querySegments.contains(KeyValue("checkpoint-properties", "eyJleGVjdXRpb25JRCI6WyJpZDEiLCJpZDIiXX0="))) + assert(r.uri.querySegments.contains(KeyValue("from", "2026-06-01T00:00:00Z"))) + assert(!r.uri.paramsMap.contains("to")) + assert(!r.uri.paramsMap.contains("latest-first")) Response.ok(checkpointsResponse) } val filter = CheckpointFilter( name = Some("Test checkpoints 1"), - properties = Map("executionID" -> Set("id1", "id2")) + properties = Map("executionID" -> Set("id1", "id2")), + from = Some(ZonedDateTime.parse("2026-06-01T00:00:00Z")) ) val result = PartitioningReader(AtumPartitions(List("a" -> "b", "c" -> "d"))).getCheckpointsPage(filter = filter) assert(result.map(_.data.size) == Right(2)) diff --git a/reader/src/test/scala/za/co/absa/atum/reader/requests/CheckpointFilterUnitTests.scala b/reader/src/test/scala/za/co/absa/atum/reader/requests/CheckpointFilterUnitTests.scala index 9c4ed8ee7..74ed24ae6 100644 --- a/reader/src/test/scala/za/co/absa/atum/reader/requests/CheckpointFilterUnitTests.scala +++ b/reader/src/test/scala/za/co/absa/atum/reader/requests/CheckpointFilterUnitTests.scala @@ -19,6 +19,8 @@ package za.co.absa.atum.reader.requests import org.scalatest.funsuite.AnyFunSuiteLike import za.co.absa.atum.model.utils.JsonSyntaxExtensions._ +import java.time.ZonedDateTime + class CheckpointFilterUnitTests extends AnyFunSuiteLike { test("An empty filter adds no query params") { @@ -43,13 +45,19 @@ class CheckpointFilterUnitTests extends AnyFunSuiteLike { ) } - test("Name and properties are combined") { - val filter = CheckpointFilter(name = Some("checkpoint name"), properties = Map("env" -> Set("prod"))) + test("The time window is sent as UTC instants and all filter parts are combined") { + val filter = CheckpointFilter( + name = Some("checkpoint name"), + from = Some(ZonedDateTime.parse("2026-06-01T02:00:00+02:00[Europe/Budapest]")), + to = Some(ZonedDateTime.parse("2026-08-01T00:00:00.5Z")), + latestFirst = Some(false) + ) assert( filter.toQueryParams == Map( "checkpoint-name" -> "checkpoint name", - // base64url-encoded {"env":"prod"} - "checkpoint-properties" -> "eyJlbnYiOiJwcm9kIn0=" + "from" -> "2026-06-01T00:00:00Z", + "to" -> "2026-08-01T00:00:00.500Z", + "latest-first" -> "false" ) ) } diff --git a/server/src/main/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpoints.scala b/server/src/main/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpoints.scala index 2534988f7..b04dc1f8b 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpoints.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpoints.scala @@ -35,6 +35,8 @@ import doobie.postgres.implicits._ import za.co.absa.atum.server.model.database.CheckpointItemWithPartitioningFromDB +import java.time.ZonedDateTime + class GetFlowCheckpoints(implicit schema: DBSchema, dbEngine: DoobieEngine[Task]) extends DoobieMultipleResultFunctionWithAggStatus[GetFlowCheckpointsArgs, Option[ CheckpointItemWithPartitioningFromDB @@ -44,7 +46,10 @@ class GetFlowCheckpoints(implicit schema: DBSchema, dbEngine: DoobieEngine[Task] fr"${input.limit}", fr"${input.offset}", fr"${input.checkpointName}", - fr"${input.checkpointProperties.map(_.asJson)}" + fr"${input.checkpointProperties.map(_.asJson)}", + fr"${input.latestFirst}", + fr"${input.from}", + fr"${input.to}" ) ) with StandardStatusHandling @@ -73,7 +78,10 @@ object GetFlowCheckpoints { limit: Int, offset: Long, checkpointName: Option[String], - checkpointProperties: Option[Map[String, Seq[String]]] + checkpointProperties: Option[Map[String, Seq[String]]], + latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime] ) val layer: URLayer[PostgresDatabaseProvider, GetFlowCheckpoints] = ZLayer { diff --git a/server/src/main/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningCheckpoints.scala b/server/src/main/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningCheckpoints.scala index b43ec3662..0632ac744 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningCheckpoints.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningCheckpoints.scala @@ -35,6 +35,8 @@ import io.circe.generic.auto._ import za.co.absa.db.fadb.status.aggregation.implementations.ByFirstRowStatusAggregator import za.co.absa.db.fadb.status.handling.implementations.StandardStatusHandling +import java.time.ZonedDateTime + class GetPartitioningCheckpoints(implicit schema: DBSchema, dbEngine: DoobieEngine[Task]) extends DoobieMultipleResultFunctionWithAggStatus[GetPartitioningCheckpointsArgs, Option[ CheckpointItemFromDB.Paginated @@ -45,7 +47,9 @@ class GetPartitioningCheckpoints(implicit schema: DBSchema, dbEngine: DoobieEngi fr"${args.offset}", fr"${args.checkpointName}", fr"${args.checkpointProperties.map(_.asJson)}", - fr"${args.latestFirst}" + fr"${args.latestFirst}", + fr"${args.from}", + fr"${args.to}" ) ) with StandardStatusHandling @@ -72,7 +76,9 @@ object GetPartitioningCheckpoints { offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], - latestFirst: Option[Boolean] + latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime] ) val layer: URLayer[PostgresDatabaseProvider, GetPartitioningCheckpoints] = ZLayer { diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/CheckpointController.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/CheckpointController.scala index 428f55a1e..4d521b999 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/CheckpointController.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/CheckpointController.scala @@ -22,6 +22,7 @@ import za.co.absa.atum.model.envelopes.SuccessResponse.{PaginatedResponse, Singl import zio.IO import zio.macros.accessible +import java.time.ZonedDateTime import java.util.UUID @accessible @@ -45,6 +46,8 @@ trait CheckpointController { checkpointName: Option[String] = None, checkpointProperties: Option[Map[String, Seq[String]]] = None, latestFirst: Option[Boolean] = None, + from: Option[ZonedDateTime] = None, + to: Option[ZonedDateTime] = None, includeProperties: Boolean ): IO[ErrorResponse, PaginatedResponse[CheckpointV2DTO]] diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/CheckpointControllerImpl.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/CheckpointControllerImpl.scala index 46897914e..464cf05ec 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/CheckpointControllerImpl.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/CheckpointControllerImpl.scala @@ -25,6 +25,7 @@ import za.co.absa.atum.server.api.v2.service.CheckpointService import za.co.absa.atum.server.model.PaginatedResult import zio._ +import java.time.ZonedDateTime import java.util.UUID class CheckpointControllerImpl(checkpointService: CheckpointService) extends CheckpointController with BaseController { @@ -66,6 +67,8 @@ class CheckpointControllerImpl(checkpointService: CheckpointService) extends Che checkpointName: Option[String] = None, checkpointProperties: Option[Map[String, Seq[String]]] = None, latestFirst: Option[Boolean] = None, + from: Option[ZonedDateTime] = None, + to: Option[ZonedDateTime] = None, includeProperties: Boolean ): IO[ErrorResponse, PaginatedResponse[CheckpointV2DTO]] = { mapToPaginatedResponse( @@ -79,6 +82,8 @@ class CheckpointControllerImpl(checkpointService: CheckpointService) extends Che checkpointName, checkpointProperties, latestFirst, + from, + to, includeProperties ) ) diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/FlowController.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/FlowController.scala index 23be150c6..90cca4761 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/FlowController.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/FlowController.scala @@ -22,6 +22,8 @@ import za.co.absa.atum.model.envelopes.SuccessResponse.PaginatedResponse import zio.IO import zio.macros.accessible +import java.time.ZonedDateTime + @accessible trait FlowController { @@ -31,6 +33,9 @@ trait FlowController { offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], + latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ): IO[ErrorResponse, PaginatedResponse[CheckpointWithPartitioningDTO]] diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/FlowControllerImpl.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/FlowControllerImpl.scala index 8a031b0f4..3d1419de0 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/FlowControllerImpl.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/controller/FlowControllerImpl.scala @@ -24,6 +24,8 @@ import za.co.absa.atum.server.api.v2.service.FlowService import za.co.absa.atum.server.model.PaginatedResult import zio._ +import java.time.ZonedDateTime + class FlowControllerImpl(flowService: FlowService) extends FlowController with BaseController { // to be replaced (and moved to checkpointcontroller) with new implementation in #233 @@ -33,11 +35,24 @@ class FlowControllerImpl(flowService: FlowService) extends FlowController with B offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], + latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ): IO[ErrorResponse, PaginatedResponse[CheckpointWithPartitioningDTO]] = { val flowData = serviceCall[PaginatedResult[CheckpointWithPartitioningDTO], PaginatedResult[CheckpointWithPartitioningDTO]]( - flowService.getFlowCheckpoints(flowId, limit, offset, checkpointName, checkpointProperties, includeProperties) + flowService.getFlowCheckpoints( + flowId, + limit, + offset, + checkpointName, + checkpointProperties, + latestFirst, + from, + to, + includeProperties + ) ) mapToPaginatedResponse(limit, offset, flowData) } diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/http/Endpoints.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/http/Endpoints.scala index 719cb657d..34db55dec 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/http/Endpoints.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/http/Endpoints.scala @@ -20,7 +20,7 @@ import sttp.model.StatusCode import sttp.tapir.generic.auto.schemaForCaseClass import sttp.tapir.json.circe.jsonBody import sttp.tapir.ztapir._ -import sttp.tapir.{Codec, CodecFormat, DecodeResult, PublicEndpoint, ValidationResult, Validator} +import sttp.tapir.{Codec, CodecFormat, DecodeResult, EndpointInput, PublicEndpoint, ValidationResult, Validator} import za.co.absa.atum.model.ApiPaths._ import za.co.absa.atum.model.dto._ import za.co.absa.atum.model.envelopes.ErrorResponse @@ -29,6 +29,7 @@ import za.co.absa.atum.model.utils.JsonSyntaxExtensions._ import za.co.absa.atum.server.api.v2.controller.{CheckpointController, FlowController, PartitioningController} import za.co.absa.atum.server.api.common.http.{BaseEndpoints, HttpEnv} +import java.time.ZonedDateTime import java.util.UUID import scala.util.{Failure, Success, Try} @@ -58,6 +59,21 @@ object Endpoints extends BaseEndpoints { ) }) + // Optional half-open window `from <= process start time < to` for checkpoint queries, supplied as ISO-8601 date-times + // with an offset, e.g. `from=2026-06-01T00:00:00Z`. A malformed value, or `from` not before `to`, results in a + // 400 Bad Request. + private val checkpointTimeWindowInput: EndpointInput[(Option[ZonedDateTime], Option[ZonedDateTime])] = + query[Option[ZonedDateTime]]("from") + .description("inclusive lower bound of the checkpoint's process start time (ISO-8601 date-time with offset)") + .and( + query[Option[ZonedDateTime]]("to") + .description("exclusive upper bound of the checkpoint's process start time (ISO-8601 date-time with offset)") + ) + .validate(Validator.custom { + case (Some(from), Some(to)) if !from.isBefore(to) => ValidationResult.Invalid("'from' must be before 'to'") + case _ => ValidationResult.Valid + }) + val postCheckpointEndpoint : PublicEndpoint[(Long, CheckpointV2DTO), ErrorResponse, (SingleSuccessResponse[CheckpointV2DTO], String), Any] = { apiV2.post @@ -126,7 +142,7 @@ object Endpoints extends BaseEndpoints { } val getPartitioningCheckpointsEndpoint - : PublicEndpoint[(Long, Int, Long, Option[String], Option[Map[String, Seq[String]]], Option[Boolean], Boolean), ErrorResponse, PaginatedResponse[ + : PublicEndpoint[(Long, Int, Long, Option[String], Option[Map[String, Seq[String]]], Option[Boolean], Option[ZonedDateTime], Option[ZonedDateTime], Boolean), ErrorResponse, PaginatedResponse[ CheckpointV2DTO ], Any] = { apiV2.get @@ -137,6 +153,7 @@ object Endpoints extends BaseEndpoints { .in(query[Option[String]]("checkpoint-name")) .in(query[Option[Map[String, Seq[String]]]]("checkpoint-properties")) .in(query[Option[Boolean]]("latest-first")) + .in(checkpointTimeWindowInput) .in(query[Boolean]("include-properties").default(false)) .out(statusCode(StatusCode.Ok)) .out(jsonBody[PaginatedResponse[CheckpointV2DTO]]) @@ -144,7 +161,7 @@ object Endpoints extends BaseEndpoints { } val getFlowCheckpointsEndpoint - : PublicEndpoint[(Long, Int, Long, Option[String], Option[Map[String, Seq[String]]], Boolean), ErrorResponse, PaginatedResponse[ + : PublicEndpoint[(Long, Int, Long, Option[String], Option[Map[String, Seq[String]]], Option[Boolean], Option[ZonedDateTime], Option[ZonedDateTime], Boolean), ErrorResponse, PaginatedResponse[ CheckpointWithPartitioningDTO ], Any] = { apiV2.get @@ -153,6 +170,8 @@ object Endpoints extends BaseEndpoints { .in(query[Long]("offset").default(0L).validate(Validator.min(0L))) .in(query[Option[String]]("checkpoint-name")) .in(query[Option[Map[String, Seq[String]]]]("checkpoint-properties")) + .in(query[Option[Boolean]]("latest-first")) + .in(checkpointTimeWindowInput) .in(query[Boolean]("include-properties").default(false)) .out(statusCode(StatusCode.Ok)) .out(jsonBody[PaginatedResponse[CheckpointWithPartitioningDTO]]) @@ -261,23 +280,23 @@ object Endpoints extends BaseEndpoints { } ), createServerEndpoint[ - (Long, Int, Long, Option[String], Option[Map[String, Seq[String]]], Option[Boolean], Boolean), + (Long, Int, Long, Option[String], Option[Map[String, Seq[String]]], Option[Boolean], Option[ZonedDateTime], Option[ZonedDateTime], Boolean), ErrorResponse, PaginatedResponse[CheckpointV2DTO] ]( getPartitioningCheckpointsEndpoint, - { case (partitioningId: Long, limit: Int, offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], latestFirst: Option[Boolean], includeProperties: Boolean) => - CheckpointController.getPartitioningCheckpoints(partitioningId, limit, offset, checkpointName, checkpointProperties, latestFirst, includeProperties) + { case (partitioningId: Long, limit: Int, offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], latestFirst: Option[Boolean], from: Option[ZonedDateTime], to: Option[ZonedDateTime], includeProperties: Boolean) => + CheckpointController.getPartitioningCheckpoints(partitioningId, limit, offset, checkpointName, checkpointProperties, latestFirst, from, to, includeProperties) } ), createServerEndpoint[ - (Long, Int, Long, Option[String], Option[Map[String, Seq[String]]], Boolean), + (Long, Int, Long, Option[String], Option[Map[String, Seq[String]]], Option[Boolean], Option[ZonedDateTime], Option[ZonedDateTime], Boolean), ErrorResponse, PaginatedResponse[CheckpointWithPartitioningDTO] ]( getFlowCheckpointsEndpoint, - { case (flowId: Long, limit: Int, offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], includeProperties: Boolean) => - FlowController.getFlowCheckpoints(flowId, limit, offset, checkpointName, checkpointProperties, includeProperties) + { case (flowId: Long, limit: Int, offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], latestFirst: Option[Boolean], from: Option[ZonedDateTime], to: Option[ZonedDateTime], includeProperties: Boolean) => + FlowController.getFlowCheckpoints(flowId, limit, offset, checkpointName, checkpointProperties, latestFirst, from, to, includeProperties) } ), createServerEndpoint(getPartitioningByIdEndpoint, PartitioningController.getPartitioningById), diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepository.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepository.scala index 14856ecf2..018cf3df6 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepository.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepository.scala @@ -22,6 +22,7 @@ import za.co.absa.atum.server.model.PaginatedResult import zio._ import zio.macros.accessible +import java.time.ZonedDateTime import java.util.UUID @accessible @@ -39,6 +40,8 @@ trait CheckpointRepository { checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ): IO[DatabaseError, PaginatedResult[CheckpointV2DTO]] } diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepositoryImpl.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepositoryImpl.scala index 3a8cb1c22..d1c13b425 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepositoryImpl.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepositoryImpl.scala @@ -31,6 +31,7 @@ import za.co.absa.atum.server.model.database.CheckpointItemFromDB import zio._ import zio.interop.catz.asyncInstance +import java.time.ZonedDateTime import java.util.UUID class CheckpointRepositoryImpl( @@ -77,11 +78,22 @@ class CheckpointRepositoryImpl( checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ): IO[DatabaseError, PaginatedResult[CheckpointV2DTO]] = { dbMultipleResultCallWithAggregatedStatus( getPartitioningCheckpointsFn( - GetPartitioningCheckpointsArgs(partitioningId, limit, offset, checkpointName, checkpointProperties, latestFirst) + GetPartitioningCheckpointsArgs( + partitioningId, + limit, + offset, + checkpointName, + checkpointProperties, + latestFirst, + from, + to + ) ), "getPartitioningCheckpoints" ) diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/FlowRepository.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/FlowRepository.scala index d47df8ff1..15e9b9951 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/FlowRepository.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/FlowRepository.scala @@ -22,6 +22,8 @@ import za.co.absa.atum.server.model.PaginatedResult import zio._ import zio.macros.accessible +import java.time.ZonedDateTime + @accessible trait FlowRepository { @@ -31,6 +33,9 @@ trait FlowRepository { offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], + latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ): IO[DatabaseError, PaginatedResult[CheckpointWithPartitioningDTO]] diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/FlowRepositoryImpl.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/FlowRepositoryImpl.scala index b668384be..d861b046d 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/FlowRepositoryImpl.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/repository/FlowRepositoryImpl.scala @@ -29,6 +29,8 @@ import za.co.absa.atum.server.model.database.CheckpointItemWithPartitioningFromD import zio._ import zio.interop.catz.asyncInstance +import java.time.ZonedDateTime + class FlowRepositoryImpl( getFlowCheckpointsFn: GetFlowCheckpoints, override val getCheckpointPropertiesFn: GetCheckpointProperties @@ -40,10 +42,15 @@ class FlowRepositoryImpl( offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], + latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ): IO[DatabaseError, PaginatedResult[CheckpointWithPartitioningDTO]] = { dbMultipleResultCallWithAggregatedStatus( - getFlowCheckpointsFn(GetFlowCheckpointsArgs(flowId, limit, offset, checkpointName, checkpointProperties)), + getFlowCheckpointsFn( + GetFlowCheckpointsArgs(flowId, limit, offset, checkpointName, checkpointProperties, latestFirst, from, to) + ), "getFlowCheckpoints" ) .map(_.flatten) diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/service/CheckpointService.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/service/CheckpointService.scala index 4e6d4439e..c59ca56cd 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/service/CheckpointService.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/service/CheckpointService.scala @@ -22,6 +22,7 @@ import za.co.absa.atum.server.model.PaginatedResult import zio.IO import zio.macros.accessible +import java.time.ZonedDateTime import java.util.UUID @accessible @@ -42,6 +43,8 @@ trait CheckpointService { checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ): IO[ServiceError, PaginatedResult[CheckpointV2DTO]] diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/service/CheckpointServiceImpl.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/service/CheckpointServiceImpl.scala index 05428e894..a53b57285 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/service/CheckpointServiceImpl.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/service/CheckpointServiceImpl.scala @@ -23,6 +23,7 @@ import za.co.absa.atum.server.api.v2.repository.CheckpointRepository import za.co.absa.atum.server.model.PaginatedResult import zio._ +import java.time.ZonedDateTime import java.util.UUID class CheckpointServiceImpl(checkpointRepository: CheckpointRepository) extends CheckpointService with BaseService { @@ -52,6 +53,8 @@ class CheckpointServiceImpl(checkpointRepository: CheckpointRepository) extends checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ): IO[ServiceError, PaginatedResult[CheckpointV2DTO]] = { repositoryCall( @@ -62,6 +65,8 @@ class CheckpointServiceImpl(checkpointRepository: CheckpointRepository) extends checkpointName, checkpointProperties, latestFirst, + from, + to, includeProperties ), "getPartitioningCheckpoints" diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/service/FlowService.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/service/FlowService.scala index 27e0bbe5c..5563086b9 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/service/FlowService.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/service/FlowService.scala @@ -22,6 +22,8 @@ import za.co.absa.atum.server.model.PaginatedResult import zio._ import zio.macros.accessible +import java.time.ZonedDateTime + @accessible trait FlowService { @@ -31,6 +33,9 @@ trait FlowService { offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], + latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ): IO[ServiceError, PaginatedResult[CheckpointWithPartitioningDTO]] diff --git a/server/src/main/scala/za/co/absa/atum/server/api/v2/service/FlowServiceImpl.scala b/server/src/main/scala/za/co/absa/atum/server/api/v2/service/FlowServiceImpl.scala index 936979a0c..555b23271 100644 --- a/server/src/main/scala/za/co/absa/atum/server/api/v2/service/FlowServiceImpl.scala +++ b/server/src/main/scala/za/co/absa/atum/server/api/v2/service/FlowServiceImpl.scala @@ -23,6 +23,8 @@ import za.co.absa.atum.server.api.v2.repository.FlowRepository import za.co.absa.atum.server.model.PaginatedResult import zio._ +import java.time.ZonedDateTime + class FlowServiceImpl(flowRepository: FlowRepository) extends FlowService with BaseService { override def getFlowCheckpoints( @@ -31,10 +33,23 @@ class FlowServiceImpl(flowRepository: FlowRepository) extends FlowService with B offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], + latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ): IO[ServiceError, PaginatedResult[CheckpointWithPartitioningDTO]] = { repositoryCall( - flowRepository.getFlowCheckpoints(flowId, limit, offset, checkpointName, checkpointProperties, includeProperties), + flowRepository.getFlowCheckpoints( + flowId, + limit, + offset, + checkpointName, + checkpointProperties, + latestFirst, + from, + to, + includeProperties + ), "getFlowCheckpoints" ) } diff --git a/server/src/main/scala/za/co/absa/atum/server/model/database/CheckpointItemWithPartitioningFromDB.scala b/server/src/main/scala/za/co/absa/atum/server/model/database/CheckpointItemWithPartitioningFromDB.scala index 25b1cf9e7..2ad4d5337 100644 --- a/server/src/main/scala/za/co/absa/atum/server/model/database/CheckpointItemWithPartitioningFromDB.scala +++ b/server/src/main/scala/za/co/absa/atum/server/model/database/CheckpointItemWithPartitioningFromDB.scala @@ -98,10 +98,8 @@ object CheckpointItemWithPartitioningFromDB { checkpointItems: Seq[CheckpointItemWithPartitioningFromDB] ): Either[Throwable, Seq[CheckpointWithPartitioningDTO]] = { val groupedItems = checkpointItems.groupBy(_.idCheckpoint) - val orderedCheckpointIds = checkpointItems - .sortBy(_.checkpointStartTime)(Ordering[ZonedDateTime].reverse) - .map(_.idCheckpoint) - .distinct + // the DB function returns the items already ordered (latest or earliest first, as requested), so keep that order + val orderedCheckpointIds = checkpointItems.map(_.idCheckpoint).distinct val result = orderedCheckpointIds.map { id => fromItemsToCheckpointWithPartitioningDTO(groupedItems(id)) } diff --git a/server/src/test/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpointsIntegrationTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpointsIntegrationTests.scala index d21a68ece..fda0905e4 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpointsIntegrationTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpointsIntegrationTests.scala @@ -97,7 +97,10 @@ object GetFlowCheckpointsIntegrationTests extends ConfigProviderTest { limit = 10, offset = 0L, checkpointName = Some("TestCheckpointName"), - checkpointProperties = None + checkpointProperties = None, + latestFirst = None, + from = None, + to = None ) for { @@ -112,7 +115,10 @@ object GetFlowCheckpointsIntegrationTests extends ConfigProviderTest { limit = 2, offset = offset, checkpointName = Some(checkpointName), - checkpointProperties = Some(Map("executionID" -> Seq("a", "b"))) + checkpointProperties = Some(Map("executionID" -> Seq("a", "b"))), + latestFirst = None, + from = None, + to = None ) ) @@ -131,6 +137,30 @@ object GetFlowCheckpointsIntegrationTests extends ConfigProviderTest { firstPage == (Seq(matching3, matching2), true), secondPage == (Seq(matching1), false) ) + }, + test("Should apply the earliest-first order and the process start time window") { + def getWindowPage(flowId: Long) = getPage( + GetFlowCheckpointsArgs( + flowId = flowId, + limit = 10, + offset = 0L, + checkpointName = None, + checkpointProperties = None, + latestFirst = Some(false), + from = Some(firstStartTime.plusMinutes(2)), + to = Some(firstStartTime.plusMinutes(4)) + ) + ) + + for { + partitioningId <- createPartitioning + flowId <- getMainFlowId(partitioningId) + _ <- writeCheckpoint(partitioningId, checkpointName, None, startMinute = 1) + atWindowStart <- writeCheckpoint(partitioningId, checkpointName, None, startMinute = 2) + inWindow <- writeCheckpoint(partitioningId, checkpointName, None, startMinute = 3) + _ <- writeCheckpoint(partitioningId, checkpointName, None, startMinute = 4) + page <- getWindowPage(flowId) + } yield assertTrue(page == (Seq(atWindowStart, inWindow), false)) } ).provide( GetFlowCheckpoints.layer, diff --git a/server/src/test/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowPartitioningsIntegrationTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowPartitioningsIntegrationTests.scala index af3a3232c..11bed2633 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowPartitioningsIntegrationTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowPartitioningsIntegrationTests.scala @@ -16,18 +16,37 @@ package za.co.absa.atum.server.api.database.flows.functions +import za.co.absa.atum.model.dto.{PartitionDTO, PartitioningSubmitV2DTO} import za.co.absa.atum.server.ConfigProviderTest import za.co.absa.atum.server.api.TestTransactorProvider import za.co.absa.atum.server.api.database.PostgresDatabaseProvider import za.co.absa.atum.server.api.database.flows.functions.GetFlowPartitionings.GetFlowPartitioningsArgs +import za.co.absa.atum.server.api.database.runs.functions.{CreatePartitioning, GetPartitioningMainFlow} import za.co.absa.db.fadb.exceptions.DataNotFoundException import za.co.absa.db.fadb.status.FunctionStatus import zio.{Scope, ZIO} import zio.test.{Spec, TestEnvironment, assertTrue} import zio.interop.catz.asyncInstance +import java.util.UUID + object GetFlowPartitioningsIntegrationTests extends ConfigProviderTest { + private def createPartitioning(parentId: Option[Long]): ZIO[CreatePartitioning, Throwable, Long] = + ZIO + .serviceWithZIO[CreatePartitioning]( + _(PartitioningSubmitV2DTO(Seq(PartitionDTO("page", UUID.randomUUID().toString)), parentId, "Pagination author")) + ) + .flatMap(ZIO.fromEither(_)) + .map(_.data) + + private def getPage(flowId: Long, offset: Long): ZIO[GetFlowPartitionings, Throwable, Seq[(Long, Boolean)]] = + ZIO + .serviceWithZIO[GetFlowPartitionings](_(GetFlowPartitioningsArgs(flowId, 2, offset))) + .flatMap(ZIO.fromEither(_)) + .flatMap(rows => ZIO.foreach(rows)(row => ZIO.getOrFail(row.data))) + .map(_.map(row => (row.id, row.hasMore))) + override def spec: Spec[Unit with TestEnvironment with Scope, Any] = { suite("GetFlowPartitioningsIntegrationTests")( test("Returns expected DataNotFoundException when flow not found") { @@ -35,12 +54,34 @@ object GetFlowPartitioningsIntegrationTests extends ConfigProviderTest { getFlowPartitionings <- ZIO.service[GetFlowPartitionings] result <- getFlowPartitionings(GetFlowPartitioningsArgs(0L, 1, 1L)) } yield assertTrue(result == Left(DataNotFoundException(FunctionStatus(41, "Flow not found")))) + }, + test("Decodes ordered partitionings and paging metadata at positive offsets") { + for { + parentId <- createPartitioning(None) + childIds <- ZIO.foreach(1 to 4)(_ => createPartitioning(Some(parentId))) + ids = (parentId +: childIds).sorted + mainFlow <- ZIO.serviceWithZIO[GetPartitioningMainFlow](_(parentId)).flatMap(ZIO.fromEither(_)) + flowId <- ZIO.getOrFail(mainFlow.data.map(_.id)) + firstPage <- getPage(flowId, 0L) + middlePage <- getPage(flowId, 2L) + lastPartialPage <- getPage(flowId, 4L) + lastFullPage <- getPage(flowId, 3L) + beyondEndPage <- getPage(flowId, 6L) + } yield assertTrue( + firstPage == ids.take(2).map(id => (id, true)), + middlePage == ids.slice(2, 4).map(id => (id, true)), + lastPartialPage == Seq((ids.last, false)), + lastFullPage == ids.takeRight(2).map(id => (id, false)), + beyondEndPage.isEmpty + ) } ) }.provide( GetFlowPartitionings.layer, + CreatePartitioning.layer, + GetPartitioningMainFlow.layer, PostgresDatabaseProvider.layer, - TestTransactorProvider.layerWithRollback + TestTransactorProvider.layerWithSingleTransactionRollback ) } diff --git a/server/src/test/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningCheckpointsIntegrationTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningCheckpointsIntegrationTests.scala index 93fb53dbe..db1d04999 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningCheckpointsIntegrationTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningCheckpointsIntegrationTests.scala @@ -87,7 +87,9 @@ object GetPartitioningCheckpointsIntegrationTests extends ConfigProviderTest { offset = 0L, checkpointName = None, checkpointProperties = None, - latestFirst = None + latestFirst = None, + from = None, + to = None ) for { @@ -103,7 +105,10 @@ object GetPartitioningCheckpointsIntegrationTests extends ConfigProviderTest { offset = offset, checkpointName = Some(checkpointName), checkpointProperties = Some(Map("executionID" -> Seq("a", "b"))), - latestFirst = latestFirst + latestFirst = latestFirst, + from = None, + to = None + ) ) @@ -123,6 +128,29 @@ object GetPartitioningCheckpointsIntegrationTests extends ConfigProviderTest { secondPage == (Seq(matching1), false), earliestFirstPage == (Seq(matching1, matching2), true) ) + }, + test("Should apply the earliest-first order and the process start time window") { + def getWindowPage(partitioningId: Long) = getPage( + GetPartitioningCheckpointsArgs( + partitioningId = partitioningId, + limit = 10, + offset = 0L, + checkpointName = None, + checkpointProperties = None, + latestFirst = Some(false), + from = Some(firstStartTime.plusMinutes(2)), + to = Some(firstStartTime.plusMinutes(4)) + ) + ) + + for { + partitioningId <- createPartitioning + _ <- writeCheckpoint(partitioningId, checkpointName, None, startMinute = 1) + atWindowStart <- writeCheckpoint(partitioningId, checkpointName, None, startMinute = 2) + inWindow <- writeCheckpoint(partitioningId, checkpointName, None, startMinute = 3) + _ <- writeCheckpoint(partitioningId, checkpointName, None, startMinute = 4) + page <- getWindowPage(partitioningId) + } yield assertTrue(page == (Seq(atWindowStart, inWindow), false)) } ).provide( GetPartitioningCheckpoints.layer, diff --git a/server/src/test/scala/za/co/absa/atum/server/api/v2/controller/CheckpointControllerUnitTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/v2/controller/CheckpointControllerUnitTests.scala index 9a846d9d5..b44674f70 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/v2/controller/CheckpointControllerUnitTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/v2/controller/CheckpointControllerUnitTests.scala @@ -51,11 +51,11 @@ object CheckpointControllerUnitTests extends ConfigProviderTest with TestData { when(checkpointServiceMock.getCheckpoint(partitioningId1, checkpointV2DTO3.id, includeProperties = false)) .thenReturn(ZIO.fail(GeneralServiceError("boom!"))) - when(checkpointServiceMock.getPartitioningCheckpoints(partitioningId1, 10, 0L, None, None, None, includeProperties = false)) + when(checkpointServiceMock.getPartitioningCheckpoints(partitioningId1, 10, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.succeed(ResultHasMore(Seq(checkpointV2DTO1)))) - when(checkpointServiceMock.getPartitioningCheckpoints(partitioningId2, 10, 0L, None, None, None, includeProperties = false)) + when(checkpointServiceMock.getPartitioningCheckpoints(partitioningId2, 10, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.succeed(ResultNoMore(Seq(checkpointV2DTO1)))) - when(checkpointServiceMock.getPartitioningCheckpoints(0L, 10, 0L, None, None, None, includeProperties = false)) + when(checkpointServiceMock.getPartitioningCheckpoints(0L, 10, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.fail(NotFoundServiceError("Partitioning not found"))) private val checkpointServiceMockLayer = ZLayer.succeed(checkpointServiceMock) diff --git a/server/src/test/scala/za/co/absa/atum/server/api/v2/controller/FlowControllerUnitTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/v2/controller/FlowControllerUnitTests.scala index d78be581e..eabe38774 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/v2/controller/FlowControllerUnitTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/v2/controller/FlowControllerUnitTests.scala @@ -29,11 +29,11 @@ import zio.test._ object FlowControllerUnitTests extends ZIOSpecDefault with TestData { private val flowServiceMock = mock(classOf[FlowService]) - when(flowServiceMock.getFlowCheckpoints(1L, 5, 2L, None, None, includeProperties = false)) + when(flowServiceMock.getFlowCheckpoints(1L, 5, 2L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.succeed(ResultHasMore(Seq(checkpointWithPartitioningDTO1)))) - when(flowServiceMock.getFlowCheckpoints(2L, 5, 0L, None, None, includeProperties = false)) + when(flowServiceMock.getFlowCheckpoints(2L, 5, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.succeed(ResultNoMore(Seq(checkpointWithPartitioningDTO2)))) - when(flowServiceMock.getFlowCheckpoints(3L, 5, 0L, None, None, includeProperties = false)) + when(flowServiceMock.getFlowCheckpoints(3L, 5, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.fail(NotFoundServiceError("Flow not found"))) private val flowServiceMockLayer = ZLayer.succeed(flowServiceMock) @@ -43,20 +43,20 @@ object FlowControllerUnitTests extends ZIOSpecDefault with TestData { suite("GetFlowCheckpointsV2Suite")( test("Returns expected Seq[CheckpointV2DTO] with Pagination indicating there is more data available") { for { - result <- FlowController.getFlowCheckpoints(1L, 5, 2L, None, None, includeProperties = false) + result <- FlowController.getFlowCheckpoints(1L, 5, 2L, None, None, None, None, None, includeProperties = false) } yield assertTrue( result.data == Seq(checkpointWithPartitioningDTO1) && result.pagination == Pagination(5, 2, hasMore = true) ) }, test("Returns expected Seq[CheckpointV2DTO] with Pagination indicating there is no more data available") { for { - result <- FlowController.getFlowCheckpoints(2L, 5, 0L, None, None, includeProperties = false) + result <- FlowController.getFlowCheckpoints(2L, 5, 0L, None, None, None, None, None, includeProperties = false) } yield assertTrue( result.data == Seq(checkpointWithPartitioningDTO2) && result.pagination == Pagination(5, 0, hasMore = false) ) }, test("Returns expected NotFoundServiceError when service returns NotFoundServiceError") { - assertZIO(FlowController.getFlowCheckpoints(3L, 5, 0L, None, None, includeProperties = false).exit)( + assertZIO(FlowController.getFlowCheckpoints(3L, 5, 0L, None, None, None, None, None, includeProperties = false).exit)( failsWithA[NotFoundErrorResponse] ) } diff --git a/server/src/test/scala/za/co/absa/atum/server/api/v2/http/GetFlowCheckpointsEndpointUnitTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/v2/http/GetFlowCheckpointsEndpointUnitTests.scala index ef56c4d42..fc5ba448e 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/v2/http/GetFlowCheckpointsEndpointUnitTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/v2/http/GetFlowCheckpointsEndpointUnitTests.scala @@ -33,6 +33,7 @@ import zio.test.Assertion.equalTo import zio.test.{Spec, TestEnvironment, ZIOSpecDefault, assertZIO} import zio.{Scope, ZIO, ZLayer} +import java.time.ZonedDateTime import java.util.UUID object GetFlowCheckpointsEndpointUnitTests extends ZIOSpecDefault with TestData { @@ -46,11 +47,11 @@ object GetFlowCheckpointsEndpointUnitTests extends ZIOSpecDefault with TestData private val encodedExecutionIdProperties = "eyJleGVjdXRpb25JRCI6IjAxOWY4OTgxLTc4NjgtNzlmYy04MWQzLTgxNDNhNDcwNmY4YSJ9" - when(flowControllerMockV2.getFlowCheckpoints(1L, 5, 0L, None, None, includeProperties = false)) + when(flowControllerMockV2.getFlowCheckpoints(1L, 5, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn( ZIO.succeed(PaginatedResponse(Seq(checkpointWithPartitioningDTO1), Pagination(5, 0, hasMore = true), uuid)) ) - when(flowControllerMockV2.getFlowCheckpoints(1L, 5, 0L, None, None, includeProperties = true)) + when(flowControllerMockV2.getFlowCheckpoints(1L, 5, 0L, None, None, None, None, None, includeProperties = true)) .thenReturn( ZIO.succeed( PaginatedResponse( @@ -60,24 +61,56 @@ object GetFlowCheckpointsEndpointUnitTests extends ZIOSpecDefault with TestData ) ) ) - when(flowControllerMockV2.getFlowCheckpoints(2L, 5, 0L, None, None, includeProperties = false)) + when(flowControllerMockV2.getFlowCheckpoints(2L, 5, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn( ZIO.succeed(PaginatedResponse(Seq(checkpointWithPartitioningDTO2), Pagination(5, 0, hasMore = false), uuid)) ) - when(flowControllerMockV2.getFlowCheckpoints(3L, 5, 0L, None, None, includeProperties = false)) + when(flowControllerMockV2.getFlowCheckpoints(3L, 5, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.fail(NotFoundErrorResponse("Flow not found for a given ID"))) when( - flowControllerMockV2.getFlowCheckpoints(1L, 5, 0L, None, Some(executionIdProperties.view.mapValues(Seq(_)).toMap), includeProperties = false) + flowControllerMockV2.getFlowCheckpoints(1L, 5, 0L, None, Some(executionIdProperties.view.mapValues(Seq(_)).toMap), None, None, None, includeProperties = false) ) .thenReturn( ZIO.succeed(PaginatedResponse(Seq(checkpointWithPartitioningDTO1), Pagination(5, 0, hasMore = true), uuid)) ) + private val windowFrom = ZonedDateTime.parse("2026-06-01T00:00:00Z") + private val windowTo = ZonedDateTime.parse("2026-08-01T00:00:00Z") + + when( + flowControllerMockV2.getFlowCheckpoints( + 1L, 5, 0L, None, None, Some(false), Some(windowFrom), Some(windowTo), includeProperties = false + ) + ) + .thenReturn( + ZIO.succeed(PaginatedResponse(Seq(checkpointWithPartitioningDTO2), Pagination(5, 0, hasMore = false), uuid)) + ) + private val flowControllerMockLayerV2 = ZLayer.succeed(flowControllerMockV2) private val getFlowCheckpointServerEndpoint = Endpoints.getFlowCheckpointsEndpoint.zServerLogic({ - case (flowId: Long, limit: Int, offset: Long, checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], includeProperties: Boolean) => - FlowController.getFlowCheckpoints(flowId, limit, offset, checkpointName, checkpointProperties, includeProperties) + case ( + flowId: Long, + limit: Int, + offset: Long, + checkpointName: Option[String], + checkpointProperties: Option[Map[String, Seq[String]]], + latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], + includeProperties: Boolean + ) => + FlowController.getFlowCheckpoints( + flowId, + limit, + offset, + checkpointName, + checkpointProperties, + latestFirst, + from, + to, + includeProperties + ) }) def spec: Spec[TestEnvironment with Scope, Any] = { @@ -175,6 +208,47 @@ object GetFlowCheckpointsEndpointUnitTests extends ZIOSpecDefault with TestData ) ) }, + test("Returns an expected PaginatedResponse[CheckpointWithPartitioningDTO] when filtering by a time window, earliest first") { + val baseUri = + uri"https://test.com/api/v2/flows/1/checkpoints?limit=5&offset=0&latest-first=false&from=2026-06-01T00:00:00Z&to=2026-08-01T00:00:00Z" + val response = basicRequest + .get(baseUri) + .response(asJson[PaginatedResponse[CheckpointWithPartitioningDTO]]) + .send(backendStub) + + val body = response.map(_.body) + val statusCode = response.map(_.code) + + assertZIO(body <&> statusCode)( + equalTo( + Right(PaginatedResponse(Seq(checkpointWithPartitioningDTO2), Pagination(5, 0, hasMore = false), uuid)), + StatusCode.Ok + ) + ) + }, + test("Returns expected 400 when 'to' is not a valid date-time") { + val baseUri = uri"https://test.com/api/v2/flows/1/checkpoints?limit=5&offset=0&to=not-a-date-time" + val response = basicRequest + .get(baseUri) + .response(asJson[PaginatedResponse[CheckpointWithPartitioningDTO]]) + .send(backendStub) + + val statusCode = response.map(_.code) + + assertZIO(statusCode)(equalTo(StatusCode.BadRequest)) + }, + test("Returns expected 400 when 'from' is after 'to'") { + val baseUri = + uri"https://test.com/api/v2/flows/1/checkpoints?limit=5&offset=0&from=2026-08-01T00:00:00Z&to=2026-06-01T00:00:00Z" + val response = basicRequest + .get(baseUri) + .response(asJson[PaginatedResponse[CheckpointWithPartitioningDTO]]) + .send(backendStub) + + val statusCode = response.map(_.code) + + assertZIO(statusCode)(equalTo(StatusCode.BadRequest)) + }, test("Returns expected 400 when checkpoint-properties is not valid base64") { val baseUri = uri"https://test.com/api/v2/flows/1/checkpoints?limit=5&offset=0&checkpoint-properties=!!!not-base64!!!" diff --git a/server/src/test/scala/za/co/absa/atum/server/api/v2/http/GetPartitioningCheckpointsEndpointUnitTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/v2/http/GetPartitioningCheckpointsEndpointUnitTests.scala index 41e320fb9..630089907 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/v2/http/GetPartitioningCheckpointsEndpointUnitTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/v2/http/GetPartitioningCheckpointsEndpointUnitTests.scala @@ -33,6 +33,7 @@ import zio.test.Assertion.equalTo import zio.test.{Spec, TestEnvironment, ZIOSpecDefault, assertZIO} import zio.{Scope, ZIO, ZLayer} +import java.time.ZonedDateTime import java.util.UUID object GetPartitioningCheckpointsEndpointUnitTests extends ZIOSpecDefault with TestData { @@ -41,9 +42,12 @@ object GetPartitioningCheckpointsEndpointUnitTests extends ZIOSpecDefault with T private val uuid = UUID.randomUUID() - when(checkpointControllerMock.getPartitioningCheckpoints(1L, 10, 0L, None, None, None, includeProperties = false)) + private val windowFrom = ZonedDateTime.parse("2026-06-01T00:00:00Z") + private val windowTo = ZonedDateTime.parse("2026-08-01T00:00:00Z") + + when(checkpointControllerMock.getPartitioningCheckpoints(1L, 10, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.succeed(PaginatedResponse(Seq(checkpointV2DTO1), Pagination(10, 0, hasMore = true), uuid))) - when(checkpointControllerMock.getPartitioningCheckpoints(1L, 10, 0L, None, None, None, includeProperties = true)) + when(checkpointControllerMock.getPartitioningCheckpoints(1L, 10, 0L, None, None, None, None, None, includeProperties = true)) .thenReturn( ZIO.succeed( PaginatedResponse( @@ -53,14 +57,20 @@ object GetPartitioningCheckpointsEndpointUnitTests extends ZIOSpecDefault with T ) ) ) - when(checkpointControllerMock.getPartitioningCheckpoints(1L, 20, 0L, None, None, None, includeProperties = false)) + when(checkpointControllerMock.getPartitioningCheckpoints(1L, 20, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.succeed(PaginatedResponse(Seq(checkpointV2DTO1), Pagination(20, 0, hasMore = false), uuid))) - when(checkpointControllerMock.getPartitioningCheckpoints(2L, 10, 0L, None, None, None, includeProperties = false)) + when(checkpointControllerMock.getPartitioningCheckpoints(2L, 10, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.fail(NotFoundErrorResponse("partitioning not found"))) - when(checkpointControllerMock.getPartitioningCheckpoints(3L, 10, 0L, None, None, None, includeProperties = false)) + when(checkpointControllerMock.getPartitioningCheckpoints(3L, 10, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.succeed(PaginatedResponse(Seq(checkpointV2DTO1), Pagination(10, 0, hasMore = true), uuid))) - when(checkpointControllerMock.getPartitioningCheckpoints(1L, 10, 0L, None, Some(Map("executionID" -> Seq("id1", "id2"))), None, includeProperties = false)) + when(checkpointControllerMock.getPartitioningCheckpoints(1L, 10, 0L, None, Some(Map("executionID" -> Seq("id1", "id2"))), None, None, None, includeProperties = false)) .thenReturn(ZIO.succeed(PaginatedResponse(Seq(checkpointV2DTO1), Pagination(10, 0, hasMore = true), uuid))) + when( + checkpointControllerMock.getPartitioningCheckpoints( + 1L, 10, 0L, None, None, None, Some(windowFrom), Some(windowTo), includeProperties = false + ) + ) + .thenReturn(ZIO.succeed(PaginatedResponse(Seq(checkpointV2DTO2), Pagination(10, 0, hasMore = false), uuid))) private val checkpointControllerMockLayer = ZLayer.succeed(checkpointControllerMock) @@ -74,6 +84,8 @@ object GetPartitioningCheckpointsEndpointUnitTests extends ZIOSpecDefault with T checkpointName: Option[String], checkpointProperties: Option[Map[String, Seq[String]]], latestFirst: Option[Boolean], + from: Option[ZonedDateTime], + to: Option[ZonedDateTime], includeProperties: Boolean ) => CheckpointController.getPartitioningCheckpoints( @@ -83,6 +95,8 @@ object GetPartitioningCheckpointsEndpointUnitTests extends ZIOSpecDefault with T checkpointName, checkpointProperties, latestFirst, + from, + to, includeProperties ) }) @@ -115,6 +129,52 @@ object GetPartitioningCheckpointsEndpointUnitTests extends ZIOSpecDefault with T ) ) }, + test("Returns an expected PaginatedResponse[CheckpointV2DTO] when filtering by a process start time window") { + val request = basicRequest + .get( + uri"https://test.com/api/v2/partitionings/1/checkpoints?limit=10&offset=0&from=2026-06-01T00:00:00Z&to=2026-08-01T00:00:00Z" + ) + .response(asJson[PaginatedResponse[CheckpointV2DTO]]) + + val response = request + .send(backendStub) + + val body = response.map(_.body) + val statusCode = response.map(_.code) + + assertZIO(body <&> statusCode)( + equalTo( + Right(PaginatedResponse(Seq(checkpointV2DTO2), Pagination(10, 0, hasMore = false), uuid)), + StatusCode.Ok + ) + ) + }, + test("Returns expected 400 when 'from' is not a valid date-time") { + val request = basicRequest + .get(uri"https://test.com/api/v2/partitionings/1/checkpoints?limit=10&offset=0&from=2026-06-01") + .response(asJson[PaginatedResponse[CheckpointV2DTO]]) + + val response = request + .send(backendStub) + + val statusCode = response.map(_.code) + + assertZIO(statusCode)(equalTo(StatusCode.BadRequest)) + }, + test("Returns expected 400 when 'from' is not before 'to'") { + val request = basicRequest + .get( + uri"https://test.com/api/v2/partitionings/1/checkpoints?limit=10&offset=0&from=2026-08-01T00:00:00Z&to=2026-08-01T00:00:00Z" + ) + .response(asJson[PaginatedResponse[CheckpointV2DTO]]) + + val response = request + .send(backendStub) + + val statusCode = response.map(_.code) + + assertZIO(statusCode)(equalTo(StatusCode.BadRequest)) + }, test("Returns an expected PaginatedResponse[CheckpointV2DTO] with more data available") { val request = basicRequest .get(uri"https://test.com/api/v2/partitionings/1/checkpoints?limit=10&offset=0") diff --git a/server/src/test/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepositoryUnitTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepositoryUnitTests.scala index af5308870..ef54fb94e 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepositoryUnitTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/v2/repository/CheckpointRepositoryUnitTests.scala @@ -62,11 +62,11 @@ object CheckpointRepositoryUnitTests extends ZIOSpecDefault with TestData { when(getCheckpointMockV2.apply(GetPartitioningCheckpointV2Args(partitioningId, checkpointV2DTO3.id))) .thenReturn(ZIO.fail(new Exception("boom!"))) - when(getPartitioningCheckpointsMock.apply(GetPartitioningCheckpointsArgs(0L, 10, 0L, None, None, None))) + when(getPartitioningCheckpointsMock.apply(GetPartitioningCheckpointsArgs(0L, 10, 0L, None, None, None, None, None))) .thenReturn(ZIO.right(Seq(Row(FunctionStatus(11, "ok"), Some(checkpointItemPaginatedFromDB))))) - when(getPartitioningCheckpointsMock.apply(GetPartitioningCheckpointsArgs(1L, 10, 0L, None, None, None))) + when(getPartitioningCheckpointsMock.apply(GetPartitioningCheckpointsArgs(1L, 10, 0L, None, None, None, None, None))) .thenReturn(ZIO.right(Seq(Row(FunctionStatus(11, "ok"), None)))) - when(getPartitioningCheckpointsMock.apply(GetPartitioningCheckpointsArgs(3L, 10, 0L, None, None, None))) + when(getPartitioningCheckpointsMock.apply(GetPartitioningCheckpointsArgs(3L, 10, 0L, None, None, None, None, None))) .thenReturn(ZIO.left(DataNotFoundException(FunctionStatus(41, "Partitioning not found")))) when(getCheckpointPropertiesMock.apply(GetCheckpointPropertiesArgs(checkpointV2DTO1.id))) @@ -137,14 +137,14 @@ object CheckpointRepositoryUnitTests extends ZIOSpecDefault with TestData { suite("GetPartitioningCheckpointsSuite")( test("Returns expected Seq without properties") { for { - result <- CheckpointRepository.getPartitioningCheckpoints(0L, 10, 0L, None, None, None, includeProperties = false) + result <- CheckpointRepository.getPartitioningCheckpoints(0L, 10, 0L, None, None, None, None, None, includeProperties = false) } yield assertTrue( result.isInstanceOf[ResultHasMore[CheckpointV2DTO]] && result.data == Seq(checkpointV2DTO2) ) }, test("Returns expected Seq with properties") { for { - result <- CheckpointRepository.getPartitioningCheckpoints(0L, 10, 0L, None, None, None, includeProperties = true) + result <- CheckpointRepository.getPartitioningCheckpoints(0L, 10, 0L, None, None, None, None, None, includeProperties = true) } yield assertTrue( result.isInstanceOf[ResultHasMore[CheckpointV2DTO]] && result.data == Seq( checkpointV2DTO2.copy(properties = Some(Map("propName1" -> "propValue1"))) @@ -153,13 +153,13 @@ object CheckpointRepositoryUnitTests extends ZIOSpecDefault with TestData { }, test("Returns expected Seq.empty") { for { - result <- CheckpointRepository.getPartitioningCheckpoints(1L, 10, 0L, None, None, None, includeProperties = false) + result <- CheckpointRepository.getPartitioningCheckpoints(1L, 10, 0L, None, None, None, None, None, includeProperties = false) } yield assertTrue( result.isInstanceOf[ResultNoMore[CheckpointV2DTO]] && result.data == Seq.empty[CheckpointV2DTO] ) }, test("Returns expected NotFoundDatabaseError") { - assertZIO(CheckpointRepository.getPartitioningCheckpoints(3L, 10, 0L, None, None, None, includeProperties = false).exit)( + assertZIO(CheckpointRepository.getPartitioningCheckpoints(3L, 10, 0L, None, None, None, None, None, includeProperties = false).exit)( failsWithA[NotFoundDatabaseError] ) } diff --git a/server/src/test/scala/za/co/absa/atum/server/api/v2/repository/FlowRepositoryUnitTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/v2/repository/FlowRepositoryUnitTests.scala index ff30f4b86..12c594385 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/v2/repository/FlowRepositoryUnitTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/v2/repository/FlowRepositoryUnitTests.scala @@ -38,7 +38,7 @@ object FlowRepositoryUnitTests extends ZIOSpecDefault with TestData { private val getFlowCheckpointsV2Mock = mock(classOf[GetFlowCheckpoints]) private val getCheckpointPropertiesMock: GetCheckpointProperties = mock(classOf[GetCheckpointProperties]) - when(getFlowCheckpointsV2Mock.apply(GetFlowCheckpointsArgs(1, 1, 1L, None, None))) + when(getFlowCheckpointsV2Mock.apply(GetFlowCheckpointsArgs(1, 1, 1L, None, None, None, None, None))) .thenReturn( ZIO.right( Seq( @@ -46,9 +46,9 @@ object FlowRepositoryUnitTests extends ZIOSpecDefault with TestData { ) ) ) - when(getFlowCheckpointsV2Mock.apply(GetFlowCheckpointsArgs(2, 1, 1L, None, None))) + when(getFlowCheckpointsV2Mock.apply(GetFlowCheckpointsArgs(2, 1, 1L, None, None, None, None, None))) .thenReturn(ZIO.right(Seq(Row(FunctionStatus(11, "success"), None)))) - when(getFlowCheckpointsV2Mock.apply(GetFlowCheckpointsArgs(3, 1, 1L, None, None))) + when(getFlowCheckpointsV2Mock.apply(GetFlowCheckpointsArgs(3, 1, 1L, None, None, None, None, None))) .thenReturn(ZIO.fail(DataNotFoundException(FunctionStatus(42, "Flow not found")))) when( @@ -67,12 +67,12 @@ object FlowRepositoryUnitTests extends ZIOSpecDefault with TestData { suite("GetFlowCheckpointsV2Suite")( test("Returns expected Right with CheckpointV2DTO without properties") { for { - result <- FlowRepository.getFlowCheckpoints(1, 1, 1L, None, None, includeProperties = false) + result <- FlowRepository.getFlowCheckpoints(1, 1, 1L, None, None, None, None, None, includeProperties = false) } yield assertTrue(result == ResultHasMore(Seq(checkpointWithPartitioningDTO1))) }, test("Returns expected Right with CheckpointV2DTO with properties") { for { - result <- FlowRepository.getFlowCheckpoints(1, 1, 1L, None, None, includeProperties = true) + result <- FlowRepository.getFlowCheckpoints(1, 1, 1L, None, None, None, None, None, includeProperties = true) } yield assertTrue( result == ResultHasMore( Seq(checkpointWithPartitioningDTO1.copy(properties = Some(Map("propName1" -> "propValue1")))) @@ -81,11 +81,11 @@ object FlowRepositoryUnitTests extends ZIOSpecDefault with TestData { }, test("Returns expected Right with CheckpointV2DTO") { for { - result <- FlowRepository.getFlowCheckpoints(2, 1, 1L, None, None, includeProperties = false) + result <- FlowRepository.getFlowCheckpoints(2, 1, 1L, None, None, None, None, None, includeProperties = false) } yield assertTrue(result == ResultNoMore(Seq.empty[CheckpointWithPartitioningDTO])) }, test("Returns expected DatabaseError") { - assertZIO(FlowRepository.getFlowCheckpoints(3, 1, 1L, None, None, includeProperties = false).exit)( + assertZIO(FlowRepository.getFlowCheckpoints(3, 1, 1L, None, None, None, None, None, includeProperties = false).exit)( failsWithA[NotFoundDatabaseError] ) } diff --git a/server/src/test/scala/za/co/absa/atum/server/api/v2/service/CheckpointServiceUnitTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/v2/service/CheckpointServiceUnitTests.scala index e2c5406dc..eaba19979 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/v2/service/CheckpointServiceUnitTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/v2/service/CheckpointServiceUnitTests.scala @@ -45,9 +45,9 @@ object CheckpointServiceUnitTests extends ZIOSpecDefault with TestData { when(checkpointRepositoryMock.getCheckpoint(partitioningId, checkpointV2DTO2.id, includeProperties = false)) .thenReturn(ZIO.fail(NotFoundDatabaseError("not found"))) - when(checkpointRepositoryMock.getPartitioningCheckpoints(1L, 10, 0L, None, None, None, includeProperties = false)) + when(checkpointRepositoryMock.getPartitioningCheckpoints(1L, 10, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.succeed(ResultHasMore(Seq(checkpointV2DTO1)))) - when(checkpointRepositoryMock.getPartitioningCheckpoints(0L, 10, 0L, None, None, None, includeProperties = false)) + when(checkpointRepositoryMock.getPartitioningCheckpoints(0L, 10, 0L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.fail(NotFoundDatabaseError("Partitioning not found"))) private val checkpointRepositoryMockLayer = ZLayer.succeed(checkpointRepositoryMock) @@ -92,13 +92,13 @@ object CheckpointServiceUnitTests extends ZIOSpecDefault with TestData { suite("GetPartitioningCheckpointsSuite")( test("Returns expected Right with Seq[CheckpointDTO]") { for { - result <- CheckpointService.getPartitioningCheckpoints(1L, 10, 0L, None, None, None, includeProperties = false) + result <- CheckpointService.getPartitioningCheckpoints(1L, 10, 0L, None, None, None, None, None, includeProperties = false) } yield assertTrue { result == ResultHasMore(Seq(checkpointV2DTO1)) } }, test("Returns expected NotFoundServiceError") { - assertZIO(CheckpointService.getPartitioningCheckpoints(0L, 10, 0L, None, None, None, includeProperties = false).exit)( + assertZIO(CheckpointService.getPartitioningCheckpoints(0L, 10, 0L, None, None, None, None, None, includeProperties = false).exit)( failsWithA[NotFoundServiceError] ) } diff --git a/server/src/test/scala/za/co/absa/atum/server/api/v2/service/FlowServiceUnitTests.scala b/server/src/test/scala/za/co/absa/atum/server/api/v2/service/FlowServiceUnitTests.scala index bcd867305..51233afbe 100644 --- a/server/src/test/scala/za/co/absa/atum/server/api/v2/service/FlowServiceUnitTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/api/v2/service/FlowServiceUnitTests.scala @@ -29,9 +29,9 @@ import zio.test._ object FlowServiceUnitTests extends ZIOSpecDefault with TestData { private val flowRepositoryMock = mock(classOf[FlowRepository]) - when(flowRepositoryMock.getFlowCheckpoints(1L, 1, 1L, None, None, includeProperties = false)) + when(flowRepositoryMock.getFlowCheckpoints(1L, 1, 1L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.succeed(ResultHasMore(Seq(checkpointWithPartitioningDTO1)))) - when(flowRepositoryMock.getFlowCheckpoints(2L, 1, 1L, None, None, includeProperties = false)) + when(flowRepositoryMock.getFlowCheckpoints(2L, 1, 1L, None, None, None, None, None, includeProperties = false)) .thenReturn(ZIO.fail(NotFoundDatabaseError("Flow not found"))) private val flowRepositoryMockLayer = ZLayer.succeed(flowRepositoryMock) @@ -42,13 +42,13 @@ object FlowServiceUnitTests extends ZIOSpecDefault with TestData { suite("GetFlowCheckpointsV2Suite")( test("Returns expected PaginatedResult[CheckpointV2DTO]") { for { - result <- FlowService.getFlowCheckpoints(1L, 1, 1L, None, None, includeProperties = false) + result <- FlowService.getFlowCheckpoints(1L, 1, 1L, None, None, None, None, None, includeProperties = false) } yield assertTrue { result == ResultHasMore(Seq(checkpointWithPartitioningDTO1)) } }, test("Returns expected ServiceError") { - assertZIO(FlowService.getFlowCheckpoints(2L, 1, 1L, None, None, includeProperties = false).exit)( + assertZIO(FlowService.getFlowCheckpoints(2L, 1, 1L, None, None, None, None, None, includeProperties = false).exit)( failsWithA[NotFoundServiceError] ) } diff --git a/server/src/test/scala/za/co/absa/atum/server/model/database/CheckpointItemWithPartitioningFromDBUnitTests.scala b/server/src/test/scala/za/co/absa/atum/server/model/database/CheckpointItemWithPartitioningFromDBUnitTests.scala index 87d2b75c6..1be1b3625 100644 --- a/server/src/test/scala/za/co/absa/atum/server/model/database/CheckpointItemWithPartitioningFromDBUnitTests.scala +++ b/server/src/test/scala/za/co/absa/atum/server/model/database/CheckpointItemWithPartitioningFromDBUnitTests.scala @@ -24,7 +24,7 @@ import java.util.UUID class CheckpointItemWithPartitioningFromDBUnitTests extends AnyFunSuiteLike with TestData { - test("groupAndConvertItemsToCheckpointWithPartitioningDTOs should group by idCheckpoint and sort by checkpointStartTime in descending order") { + test("groupAndConvertItemsToCheckpointWithPartitioningDTOs should group by idCheckpoint and keep the order of the DB rows") { val checkpointItem1 = CheckpointItemWithPartitioningFromDB( idCheckpoint = UUID.randomUUID(), checkpointName = "checkpoint1", @@ -57,14 +57,20 @@ class CheckpointItemWithPartitioningFromDBUnitTests extends AnyFunSuiteLike with hasMore = true ) - val checkpointItems = Seq(checkpointItem1, checkpointItem2) + // a second measure of the first checkpoint, i.e. another DB row of the same checkpoint + val checkpointItem1OtherMeasure = checkpointItem1.copy(measureName = "measure3", measuredColumns = Seq("col3")) - val result = CheckpointItemWithPartitioningFromDB.groupAndConvertItemsToCheckpointWithPartitioningDTOs(checkpointItems) + // earliest first, as returned by the DB for latest-first = false + val earliestFirst = CheckpointItemWithPartitioningFromDB + .groupAndConvertItemsToCheckpointWithPartitioningDTOs(Seq(checkpointItem1, checkpointItem1OtherMeasure, checkpointItem2)) + assert(earliestFirst.isRight) + assert(earliestFirst.toOption.get.map(_.id) == Seq(checkpointItem1.idCheckpoint, checkpointItem2.idCheckpoint)) + assert(earliestFirst.toOption.get.head.measurements.size == 2) - assert(result.isRight) - val checkpoints = result.toOption.get - - assert(checkpoints.size == 2) - assert(checkpoints.head.processStartTime.isAfter(checkpoints(1).processStartTime)) + // latest first, as returned by the DB by default + val latestFirst = CheckpointItemWithPartitioningFromDB + .groupAndConvertItemsToCheckpointWithPartitioningDTOs(Seq(checkpointItem2, checkpointItem1, checkpointItem1OtherMeasure)) + assert(latestFirst.isRight) + assert(latestFirst.toOption.get.map(_.id) == Seq(checkpointItem2.idCheckpoint, checkpointItem1.idCheckpoint)) } }