Doobie multiple results with aggregated status: 219 #220

TebaleloS · 2024-07-10T11:55:11Z

Release notes:

Updated fa-db library to the new version.
Refactored the db call classes according to the implementation of fa-db.
Modified the db call classes to include the aggregated status.
Refactored PartitioningRepositoryImpl class.
Refactored BaseRepository.
Added new methods to implement the db call with status for both return of single and multiple results.
Added new private methods to extract the duplicate implementation in the base repository class
Modified PartitioningRepositoryUnitTest class mocks and test cases.
Modified CreateOrUpdateAdditionalDataRepositoryImpl and CreateOrUpdateAdditionalDataRepositoryUnitTest class mocks and test cases.
Modified CheckpointRepositoryUnitTest class mocks and test cases.
Modified FlowRepositoryUnitTest class mocks and test cases.
Modified PartitioningServiceImpl class and PartitioningServiceUnit test.
Modified CreateOrUpdateAdditionalDataServiceUnitTest class mocks and test cases.

…ithAggStatus # Conflicts: # project/Dependencies.scala # server/src/test/scala/za/co/absa/atum/server/api/controller/CheckpointControllerUnitTests.scala

TebaleloS · 2024-07-26T10:19:51Z

Release notes:

Updated fa-db library to the new version.
Refactored the db call classes according to the implementation of fa-db.
Modified the db call classes to include the aggregated status.
Refactored PartitioningRepositoryImpl class.
Refactored BaseRepository.
Added new methods to implement the db call with status for both return of single and multiple results.
Added new private methods to extract the duplicate implementation in the base repository class
Modified PartitioningRepositoryUnitTest class mocks and test cases.
Modified CreateOrUpdateAdditionalDataRepositoryImpl and CreateOrUpdateAdditionalDataRepositoryUnitTest class mocks and test cases.
Modified CheckpointRepositoryUnitTest class mocks and test cases.
Modified FlowRepositoryUnitTest class mocks and test cases.
Modified PartitioningServiceImpl class and PartitioningServiceUnit test.
Modified CreateOrUpdateAdditionalDataServiceUnitTest class mocks and test cases.

salamonpavel · 2024-07-31T06:28:07Z

server/src/main/scala/za/co/absa/atum/server/api/controller/CheckpointControllerImpl.scala

serviceCallWithStatus not used

Okay, will remove the method

salamonpavel · 2024-07-31T06:31:06Z

server/src/main/scala/za/co/absa/atum/server/api/database/DoobieImplicits.scala

Most of these instances were placed in fa-db. Therefore, there is no need to keep them in Atum anymore.

salamonpavel · 2024-07-31T06:34:42Z

.../src/main/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpoints.scala

 import za.co.absa.atum.server.api.database.DoobieImplicits.Sequence.get
-
 import doobie.postgres.implicits._
 import doobie.postgres.circe.jsonb.implicits.jsonbPut
 import doobie.postgres.circe.json.implicits.jsonGet


you read jsonb, not json ... better to use jsonbGet

salamonpavel · 2024-07-31T06:38:33Z

.../src/main/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpoints.scala

 import doobie.postgres.implicits._
 import doobie.postgres.circe.jsonb.implicits.jsonbPut
 import doobie.postgres.circe.json.implicits.jsonGet
 import io.circe.syntax.EncoderOps
+import za.co.absa.db.fadb.status.aggregation.implementations.ByFirstErrorStatusAggregator
+import za.co.absa.db.fadb.status.handling.implementations.StandardStatusHandling

 class GetFlowCheckpoints(implicit schema: DBSchema, dbEngine: DoobieEngine[Task])


Tests for this class are missing.

salamonpavel · 2024-07-31T06:47:01Z

server/src/main/scala/za/co/absa/atum/server/api/database/runs/functions/WriteCheckpoint.scala

use the implicits from fa-db

salamonpavel · 2024-07-31T06:52:52Z

server/src/main/scala/za/co/absa/atum/server/api/repository/BaseRepository.scala

Please make sure formatting is applied before pushing to Github.

salamonpavel · 2024-07-31T07:00:44Z

server/src/main/scala/za/co/absa/atum/server/api/repository/BaseRepository.scala

+      DatabaseError(s"Operation '$operationName' failed with unexpected error: ${error.getMessage}")
+  }
+
+  private def dbCall[R](dbFuncCall: Task[R], operationName: String):


private def dbCall[R](dbFuncCall: Task[R], operationName: String): IO[DatabaseError, R] = { dbFuncCall .mapError(error => DatabaseError(error.getMessage)) .tapBoth( error => ZIO.logError(s"Operation '$operationName' failed: ${error.getMessage}"), _ => ZIO.logDebug(s"Operation '$operationName' succeeded in database") ) }

salamonpavel · 2024-07-31T07:05:43Z

server/src/main/scala/za/co/absa/atum/server/api/repository/BaseRepository.scala

dbSingleResultCall, dbMultipleResultCall not used

salamonpavel · 2024-07-31T07:07:32Z

server/src/main/scala/za/co/absa/atum/server/api/repository/BaseRepository.scala

+// Seq[FailedOrRow[R]] ~ Seq[Either[StatusException, Row[R]]] - dbMultipleResultCallWithStatus => IO[DatabaseError, Seq[R]]
+// FailedOrRows[R] ~ Either[StatusException, Seq[Row[R]]] - dbMultipleResultCallWithAggregatedStatus => IO[DatabaseError, Seq[R]]
+
+sealed trait PaginatedResult[R]


Why are these classes defined in this file?

Removed along with the pagination function

salamonpavel · 2024-07-31T07:08:04Z

server/src/main/scala/za/co/absa/atum/server/api/repository/BaseRepository.scala

 import zio._

+// R - dbSingleResultCall[R] => IO[DatabaseError, R]


Are these comments still needed?

salamonpavel · 2024-07-31T07:16:14Z

server/src/main/scala/za/co/absa/atum/server/api/repository/BaseRepository.scala

+    }
+  }
+
+  private def defaultErrorHandler[R](operationName: String):


type parameter is not needed

salamonpavel · 2024-07-31T07:23:08Z

server/src/main/scala/za/co/absa/atum/server/api/repository/BaseRepository.scala

-    dbFuncCall: Task[R],
-    operationName: String
-  ): IO[DatabaseError, R] = {
+  private def logOperationResult[R](operationName: String, dbFuncCall: Task[R]):


I am not sure of this. You apply partial function in the tap call assuming that the result will be Either[StatusException, R]. But the signature doesn't guarantee that, it's unsafe.

Okay, will fix that

salamonpavel · 2024-07-31T07:35:22Z

server/src/main/scala/za/co/absa/atum/server/api/repository/BaseRepository.scala

+      .flatMap {
+        case Left(statusException) => ZIO.fail(statusException)
+        case Right(value) => ZIO.succeed(
+          if (value.nonEmpty && value.head.functionStatus.statusCode == 11){


Are you sure this particular status code we want to use for "has more" cases? I think other codes were discussed in out Teams chat. Check this with others ....

This operation is removed as it is not part of this PR. But I will still go back to the discussion, and also happy to confirm with others. Thank you.

salamonpavel · 2024-07-31T07:37:50Z

server/src/main/scala/za/co/absa/atum/server/api/repository/FlowRepository.scala

@@ -24,5 +24,6 @@ import zio.macros.accessible

 @accessible
 trait FlowRepository {
-  def getFlowCheckpoints(checkpointQueryDTO: CheckpointQueryDTO): IO[DatabaseError, Seq[CheckpointFromDB]]
+  def getFlowCheckpoints(checkpointQueryDTO: CheckpointQueryDTO):


Why can't this be on a single line?

salamonpavel · 2024-07-31T07:38:10Z

server/src/main/scala/za/co/absa/atum/server/api/repository/FlowRepositoryImpl.scala

-  override def getFlowCheckpoints(checkpointQueryDTO: CheckpointQueryDTO): IO[DatabaseError, Seq[CheckpointFromDB]] = {
-    dbCall(getFlowCheckpointsFn(checkpointQueryDTO), "getFlowCheckpoints")
+  override def getFlowCheckpoints(checkpointQueryDTO: CheckpointQueryDTO):
+  IO[DatabaseError, Seq[CheckpointFromDB]] = {


formatting is off

salamonpavel · 2024-08-01T08:03:48Z

model/src/main/scala/za/co/absa/atum/model/dto/package.scala

@@ -28,4 +27,7 @@ package object dto {
  implicit val decodeAdditionalDataDTO: Decoder[AdditionalDataDTO] = Decoder.decodeMap[String, Option[String]]
  implicit val encodeAdditionalDataDTO: Encoder[AdditionalDataDTO] = Encoder.encodeMap[String, Option[String]]

+  // Implicit encoders and decoders for PartitioningDTO


I think it's not the best practice to define such implicits in package object as by doing that you implicitly import them into every possible scope in the entire package which might leaad to unintended implicit conversions or resolutions. These types should be changed. AdditionalDataDTO will undergo some changes when working on endpoints related to AdditionalData. PartitioningDTO could be modeled as case class and json serde defined in its companion or kept as is but the serde defined elsewhere.

salamonpavel · 2024-08-01T08:14:43Z

...za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpointsIntegrationTests.scala

+        for {
+          getFlowCheckpoints <- ZIO.service[GetFlowCheckpoints]
+          exit <- getFlowCheckpoints(partitioningQueryDTO).exit
+        } yield assert(exit)(failsWithA[doobie.util.invariant.NonNullableColumnRead])


The database call should never end up in NonNullableColumnRead exception. That signals nothing but incorrect implementation of the fa-db class. As of now (fa-db 0.5.0) all columns that are potentially returned with NULL value have to be modeled as Option.

salamonpavel · 2024-08-01T08:23:58Z

.../scala/za/co/absa/atum/server/api/database/runs/functions/CreateOrUpdateAdditionalData.scala

 import io.circe.syntax._

 import doobie.postgres.implicits._
-import doobie.postgres.circe.jsonb.implicits.jsonbPut
+import za.co.absa.db.fadb.doobie.postgres.circe.implicits.jsonbPut


Yep. Let's use implicits from fa-db. Btw since the doobie-postgres-circe has been made a dependency of doobie's module in fa-db there is no need to explicitly build the project with that dependency.
In other words the below dependency is not needed anymore as it's brought in transitively by the fa-db's doobie module. Pls remove it.
lazy val pgCirceDoobie = "org.tpolecat" %% "doobie-postgres-circe" % "1.0.0-RC2"

salamonpavel · 2024-08-01T08:31:58Z

.../main/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningMeasures.scala

-import za.co.absa.fadb.doobie.DoobieFunction.DoobieMultipleResultFunction
+import za.co.absa.db.fadb.DBSchema
+import za.co.absa.db.fadb.doobie.DoobieEngine
+import za.co.absa.db.fadb.doobie.DoobieFunction.DoobieMultipleResultFunctionWithAggStatus
 import za.co.absa.atum.server.api.database.PostgresDatabaseProvider
 import za.co.absa.atum.server.api.database.runs.Runs
 import zio._
 import zio.interop.catz._


I don't think this one is needed.

Missed that

salamonpavel · 2024-08-01T08:43:41Z

server/src/main/scala/za/co/absa/atum/server/api/repository/BaseRepository.scala

          )
        case Right(_) => ZIO.logDebug(s"Operation '$operationName' succeeded in database")
+        case _ => ZIO.logError(s"Operation '$operationName' did not return an Either")


It does not make much sense to me. You should require the effects on its input to be Task[FailedOrRow[R]] or Task[FailedOrRows[R]] (another method).

salamonpavel · 2024-08-01T08:56:26Z

server/src/main/scala/za/co/absa/atum/server/api/service/BaseService.scala

-    repositoryCall: IO[DatabaseError, Either[StatusException, R]],
-    operationName: String
-  ): IO[ServiceError, Either[StatusException, R]] = {
+  def repositoryCallWithStatus[R](repositoryCall: IO[DatabaseError, Either[StatusException, R]], operationName: String


Not used since we decided to handle StatusException(s) on repository level. Now the question is whether we will always propagate StatusExceptions to upper layers (Controller) in cases when we really need to (for instance when we will need to decide on resulting status code of given REST call based on the status of the database call) or we will create a hierarchy of ServiceError(s) indicating what happened on the repository level and propagate that information this way using ZIO's error channel.

salamonpavel · 2024-08-02T07:11:55Z

model/src/main/scala/za/co/absa/atum/model/dto/package.scala

 import io.circe._

 package object dto {
  type PartitioningDTO = Seq[PartitionDTO]
  type AdditionalDataDTO = Map[String, Option[String]]

+  // Todo. This implicit definition should not be defined here, so it is to be addressed in PR#221


I think the PRs actually can have different numbers than the issues. But anyway I will be taking the 221 ticket and as very next one so let's not worry about correct wording here.

salamonpavel · 2024-08-02T07:12:41Z

...ala/za/co/absa/atum/server/api/database/runs/functions/WriteCheckpointIntegrationTests.scala

 import zio.test._

 import java.time.ZonedDateTime
 import java.util.UUID

-object WriteCheckpointIntegrationTests extends ConfigProviderTest {
+object


remove the new line after object keyword

benedeki

If the seemingly unused imports are actually needed, please add comments to them.

.../src/main/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpoints.scala

benedeki · 2024-08-02T11:37:29Z

.../src/main/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowCheckpoints.scala

+    "status",
+    "status_text",


Better approach:
While not incorrect, better approach, and enuring better compatibility, is to use the
super.fieldsToSelect ++ instead of naming explicitly naming "status", "status_text",

benedeki · 2024-08-02T11:40:17Z

.../main/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningMeasures.scala

-                  $partitioningJson
-                ) ${Fragment.const(alias)};"""
-  }
+  override val fieldsToSelect: Seq[String] = Seq("status", "status_text", "measure_name", "measured_columns")


Ditto with super.fieldsToSelect.

benedeki · 2024-08-02T11:43:05Z

...in/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningCheckpoints.scala

  override val fieldsToSelect: Seq[String] = Seq(
+    "status",


Ditto with super.fieldsToSelect.

benedeki · 2024-08-02T11:44:05Z

...scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningAdditionalData.scala

+  extends DoobieMultipleResultFunctionWithAggStatus[PartitioningDTO, AdditionalDataFromDB, Task](
+    values => Seq(fr"${PartitioningForDB.fromSeqPartitionDTO(values).asJson}"))
+    with StandardStatusHandling with ByFirstErrorStatusAggregator {
+      override val fieldsToSelect: Seq[String] = Seq("status", "status_text", "ad_name", "ad_value")


Ditto with super.fieldsToSelect.

...scala/za/co/absa/atum/server/api/database/runs/functions/CreatePartitioningIfNotExists.scala

...scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningAdditionalData.scala

benedeki · 2024-08-02T12:44:15Z

...in/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningCheckpoints.scala

 import doobie.postgres.implicits._
-import doobie.postgres.circe.jsonb.implicits.jsonbPut
-import doobie.postgres.circe.json.implicits.jsonGet
+import za.co.absa.db.fadb.doobie.postgres.circe.implicits.{jsonbGet, jsonbPut}


Seems unused

JsonbGet is not needed, JsonbPut is needed.

.../main/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningMeasures.scala

benedeki · 2024-08-02T12:51:04Z

server/src/main/scala/za/co/absa/atum/server/api/database/runs/functions/WriteCheckpoint.scala

 import za.co.absa.atum.model.dto.MeasureResultDTO._
 import za.co.absa.atum.server.api.database.DoobieImplicits.Sequence.get
-import za.co.absa.atum.server.api.database.DoobieImplicits.Jsonb.jsonbArrayPut
-import doobie.postgres.circe.jsonb.implicits.jsonbGet
-import doobie.postgres.circe.jsonb.implicits.jsonbPut
+import za.co.absa.db.fadb.doobie.postgres.circe.implicits.jsonbPut
+import za.co.absa.db.fadb.doobie.postgres.circe.implicits.jsonbArrayPut
 import doobie.postgres.implicits._


Seems unused.

import za.co.absa.atum.model.dto.MeasureResultDTO._ is not needed, the rest is.

Implemented suggestions

benedeki

LGTM

Update fa-db and associated imports

c582b79

TebaleloS self-assigned this Jul 10, 2024

TebaleloS changed the title ~~Doobie multiple results with aggregated status~~ Doobie multiple results with aggregated status: 219 Jul 10, 2024

TebaleloS added Server Issues touching the server part of the project refactoring Improving code quality, paying off tech debt, aligning API, cleanup of unused code work in progress Work on this item is not yet finished (mainly intended for PRs) labels Jul 10, 2024

TebaleloS added 23 commits July 11, 2024 22:19

Merge branch 'master' into feature/#219-DoobieMultipleResultFunctionW…

da3f9f6

…ithAggStatus # Conflicts: # project/Dependencies.scala # server/src/test/scala/za/co/absa/atum/server/api/controller/CheckpointControllerUnitTests.scala

Implicit encoders and decoders for PartitioningDTO

a85e672

adding value parameter toFragmentsSeq

2de71dd

modifying the dbFunction call classes and PartitioningRepository

0176678

Modifying getFlowCheckpoints

6a3474f

Modifying partitioning service function signatures

1f6b20a

partitioning service

b1bbd32

Refactoring

166f506

Refactoring repository services

953a99a

refactoring serviceCallWithStatus in controller

075c621

incorporating statusCode dbCallWithStatus

933688b

refactoring partitioning and test cases

d70a8c1

Fixing test cases

c5dc61f

Fixing PartitioningRepositoryUnitTests

b31d9a5

Fixing PartitioningRepositoryUnitTests

ee594e4

Fixed PartitioningRepositoryUnitTests

d684420

Fixed WriteCheckpointRepositoryUnitTests

1908297

Implement defaultErrorHandler method

fa6e076

optimising partitioningRepository base repository

d51b81d

Fixing CheckpointServiceUnitTests

bff113b

Fixing PartitioningServiceUnitTests

dce3de7

remove Unit and unused imports from PartitioningServiceUnitTests

d53a58a

remove Unit and unused imports from CheckpointServiceUnitTests

1a4bfb1

salamonpavel reviewed Jul 31, 2024

View reviewed changes

Addressing GitHub comments

bb2b202

salamonpavel reviewed Aug 1, 2024

View reviewed changes

TebaleloS added 2 commits August 1, 2024 15:20

Re-addressing GitHub comments

f02bf72

Addind Todo comment

c96e90a

salamonpavel reviewed Aug 2, 2024

View reviewed changes

salamonpavel previously approved these changes Aug 2, 2024

View reviewed changes

Fixing format comments

3005404

TebaleloS dismissed salamonpavel’s stale review via 3005404 August 2, 2024 07:44

salamonpavel previously approved these changes Aug 2, 2024

View reviewed changes

benedeki reviewed Aug 2, 2024

View reviewed changes

Fixing format comments

94750ce

Implemented suggestions

TebaleloS dismissed salamonpavel’s stale review via 94750ce August 4, 2024 05:52

benedeki approved these changes Aug 5, 2024

View reviewed changes

TebaleloS merged commit 2b582d7 into master Aug 5, 2024
10 checks passed

TebaleloS deleted the feature/#219-DoobieMultipleResultFunctionWithAggStatus branch August 5, 2024 06:56

		import zio._

		// R - dbSingleResultCall[R] => IO[DatabaseError, R]

Doobie multiple results with aggregated status: 219 #220

Doobie multiple results with aggregated status: 219 #220

Conversation

TebaleloS commented Jul 10, 2024 • edited by miroslavpojer Loading

TebaleloS commented Jul 26, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

salamonpavel Jul 31, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

salamonpavel Jul 31, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

salamonpavel Aug 1, 2024 • edited Loading

Choose a reason for hiding this comment

salamonpavel Aug 1, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

salamonpavel Aug 2, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benedeki left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benedeki left a comment

Choose a reason for hiding this comment

TebaleloS commented Jul 10, 2024 •

edited by miroslavpojer

Loading

salamonpavel Jul 31, 2024 •

edited

Loading

salamonpavel Jul 31, 2024 •

edited

Loading

salamonpavel Aug 1, 2024 •

edited

Loading

salamonpavel Aug 1, 2024 •

edited

Loading

salamonpavel Aug 2, 2024 •

edited

Loading