Get Atum Context, by getting existing partitioning from DB - retrieve Measures and AdditionalData for a given partitioning #120

database/src/main/postgres/flows/V1.7.4__flows.alter.ddl

@@ -13,6 +13,7 @@
  * limitations under the License.
  */
 
-ALTER COLUMN fk_primary_partitioning SET NOT NULL;
+ALTER TABLE flows.flows
+    ALTER COLUMN fk_primary_partitioning SET NOT NULL;
 
 CREATE UNIQUE INDEX  IF NOT EXISTS unq_flows ON flows.flows (fk_primary_partitioning);

database/src/main/postgres/runs/V1.8.1__get_partitioning_measures.sql

@@ -0,0 +1,70 @@
+/*
+ * 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 runs.get_partitioning_measures(
+     IN i_partitioning          JSONB,
+     OUT status                 INTEGER,
+     OUT status_text            TEXT,
+     OUT measure_name           TEXT,
+     OUT measured_columns       TEXT[]
+ ) RETURNS SETOF record AS
+$$
+-------------------------------------------------------------------------------
+--
+-- Function: runs.get_partitioning_measures(1)
+--      Returns measures for the given partitioning
+--
+-- Parameters:
+--      i_partitioning      - partitioning we are asking the measures for
+--
+-- Returns:
+--      status              - Status code
+--      status_text         - Status message
+--      measure_name        - Name of the measure
+--      measured_columns    - Array of columns associated with the measure
+
+-- Status codes:
+--      11 - OK
+--      41 - Partitioning not found
+--
+-------------------------------------------------------------------------------
+
+DECLARE
+    _fk_partitioning                    BIGINT;
+BEGIN
+    _fk_partitioning = runs._get_id_partitioning(i_partitioning);
+
+    IF _fk_partitioning IS NULL THEN
+        status := 41;
+        status_text := 'The partitioning does not exist.';
+        RETURN NEXT;
+        RETURN;
+    END IF;
+
+    status := 11;
+    status_text := 'OK';
+
+    RETURN QUERY
+    SELECT status, status_text, md.measure_name, md.measured_columns
+    FROM runs.measure_definitions AS md
+    WHERE md.fk_partitioning = _fk_partitioning;
+
+END;
+$$
+LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
+
+ALTER FUNCTION runs.get_partitioning_measures(JSONB) OWNER TO atum_owner;
+GRANT EXECUTE ON FUNCTION runs.get_partitioning_measures(JSONB) TO atum_user;

database/src/main/postgres/runs/V1.8.2__get_partitioning_additional_data.sql

@@ -0,0 +1,71 @@
+/*
+ * 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 runs.get_partitioning_additional_data(
+    IN  i_partitioning          JSONB,
+    OUT status                  INTEGER,
+    OUT status_text             TEXT,
+    OUT ad_name                 TEXT,
+    OUT ad_value                TEXT
+) RETURNS SETOF record AS
+
+$$
+-------------------------------------------------------------------------------
+--
+-- Function: runs.get_partitioning_additional_data(1)
+--      Returns additional data for the given partitioning
+--
+-- Parameters:
+--      i_partitioning      - partitioning for requested additional data
+--
+-- Returns:
+--      status              - Status code
+--      status_text         - Status message
+--      ad_name             - Name of the additional data
+--      ad_value            - Value of the additional data
+--
+-- Status codes:
+--      11 - OK
+--      41 - Partitioning not found
+--
+-------------------------------------------------------------------------------
+
+DECLARE
+    _fk_partitioning                    BIGINT;
+BEGIN
+    _fk_partitioning = runs._get_id_partitioning(i_partitioning);
+
+    IF _fk_partitioning IS NULL THEN
+        status := 41;
+        status_text := 'The partitioning does not exist.';
+        RETURN NEXT;
+        RETURN;
+    END IF;
+
+    status = 11;
+    status_text = 'OK';
+
+    RETURN QUERY
+    SELECT status, status_text, ad.ad_name, ad.ad_value
+    FROM runs.additional_data AS ad
+    WHERE ad.fk_partitioning = _fk_partitioning;
+
+END;
+$$
+LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
+
+ALTER FUNCTION runs.get_partitioning_additional_data(JSONB) OWNER TO atum_owner;
+GRANT EXECUTE ON FUNCTION runs.get_partitioning_additional_data(JSONB) TO atum_user;

database/src/test/scala/za/co/absa/atum/database/runs/GetPartitioningAdditionalDataTest.scala

@@ -72,23 +72,20 @@ class GetPartitioningAdditionalDataTest extends DBTestSuite{
         .add("created_by", "Joseph")
         .add("ad_name", "ad_1")
         .add("ad_value", "This is the additional data for Joseph")
-        .add("updated_by", "Joseph")
     )
 
     table("runs.additional_data").insert(
       add("fk_partitioning", fkPartitioning1)
         .add("created_by", "Joseph")
         .add("ad_name", "ad_2")
         .add("ad_value", "This is the additional data for Joseph")
-        .add("updated_by", "Joseph")
     )
 
     table("runs.additional_data").insert(
       add("fk_partitioning", fkPartitioning2)
         .add("created_by", "Daniel")
         .add("ad_name", "ad_3")
         .add("ad_value", "This is the additional data for Daniel")
-        .add("updated_by", "Daniel")
     )
 
     function(fncGetPartitioningAdditionalData)
@@ -130,10 +127,6 @@ class GetPartitioningAdditionalDataTest extends DBTestSuite{
     function(fncGetPartitioningAdditionalData)
       .setParam("i_partitioning", partitioning2)
       .execute { queryResult =>
-        val results = queryResult.next()
-        assert(results.getInt("status").contains(16))
-        assert(results.getString("status_text").contains("No additional data found for the given partitioning."))
-        assert(results.getString("ad_name").isEmpty)
         assert(!queryResult.hasNext)
       }
 

database/src/test/scala/za/co/absa/atum/database/runs/GetPartitioningMeasuresTest.scala

@@ -135,9 +135,6 @@ class GetPartitioningMeasuresTest extends DBTestSuite {
     function(fncGetPartitioningMeasures)
       .setParam("i_partitioning", partitioning)
       .execute { queryResult =>
-        val results = queryResult.next()
-        assert(results.getInt("status").contains(16))
-        assert(results.getString("status_text").contains("No measures found for the given partitioning."))
         assert(!queryResult.hasNext)
       }
   }

server/src/main/scala/za/co/absa/atum/server/Main.scala

@@ -47,6 +47,8 @@ object Main extends ZIOAppDefault with Server {
           PartitioningRepositoryImpl.layer,
           CheckpointRepositoryImpl.layer,
           CreatePartitioningIfNotExists.layer,
+          GetPartitioningMeasures.layer,
+          GetPartitioningAdditionalData.layer,
           CreateOrUpdateAdditionalData.layer,
           WriteCheckpoint.layer,
           PostgresDatabaseProvider.layer,

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

@@ -23,6 +23,6 @@ import zio.macros.accessible
 
 @accessible
 trait PartitioningController {
-  def createPartitioningIfNotExists(partitioning: PartitioningSubmitDTO): IO[ErrorResponse, AtumContextDTO]
+  def createPartitioningIfNotExists(partitioningSubmitDTO: PartitioningSubmitDTO): IO[ErrorResponse, AtumContextDTO]
   def createOrUpdateAdditionalData(additionalData: AdditionalDataSubmitDTO): IO[ErrorResponse, AdditionalDataSubmitDTO]
 }

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

@@ -16,23 +16,29 @@
 
 package za.co.absa.atum.server.api.controller
 
-import za.co.absa.atum.model.dto._
+
+import za.co.absa.atum.model.dto.{AdditionalDataSubmitDTO, AtumContextDTO, PartitioningSubmitDTO}
+import za.co.absa.atum.server.api.exception.ServiceError
 import za.co.absa.atum.server.api.service.PartitioningService
-import za.co.absa.atum.server.model.ErrorResponse
+import za.co.absa.atum.server.model.{ErrorResponse, InternalServerErrorResponse}
 import zio._
 
 class PartitioningControllerImpl(partitioningService: PartitioningService)
   extends PartitioningController with BaseController {
 
-  override def createPartitioningIfNotExists(partitioning: PartitioningSubmitDTO): IO[ErrorResponse, AtumContextDTO] = {
-    serviceCallWithStatus[Unit, AtumContextDTO](
-      partitioningService.createPartitioningIfNotExists(partitioning),
-      _ => {
-        val measures: Set[MeasureDTO] = Set(MeasureDTO("count", Seq("*")))
-        val additionalData: AdditionalDataDTO = Map.empty
-        AtumContextDTO(partitioning.partitioning, measures, additionalData)
+  override def createPartitioningIfNotExists(partitioningSubmitDTO: PartitioningSubmitDTO): IO[ErrorResponse, AtumContextDTO] = {
+    for {
+      _ <- partitioningService.createPartitioningIfNotExists(partitioningSubmitDTO)
+        .mapError(serviceError => InternalServerErrorResponse(serviceError.message))
+      measures <- partitioningService.getPartitioningMeasures(partitioningSubmitDTO.partitioning)
+        .mapError {
+          serviceError: ServiceError => InternalServerErrorResponse(serviceError.message)
       }
-    )
+      additionalData <- partitioningService.getPartitioningAdditionalData(partitioningSubmitDTO.partitioning)
+        .mapError {
+          serviceError: ServiceError => InternalServerErrorResponse(serviceError.message)
+      }
+    } yield AtumContextDTO(partitioningSubmitDTO.partitioning, measures.toSet, additionalData)
   }
 
   override def createOrUpdateAdditionalData(

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

@@ -32,6 +32,9 @@ object DoobieImplicits {
   private implicit val showPGobject: Show[PGobject] = Show.show(_.getValue.take(250))
   private implicit val showPgArray: Show[PgArray] = Show.fromToString
 
+  implicit val getMapWithOptionStringValues: Get[Map[String, Option[String]]] = Get[Map[String, String]]
+      .tmap(map => map.map { case (k, v) => k -> Option(v) })
+
   object Sequence {
 
     implicit val get: Get[Seq[String]] = Get[List[String]].map(_.toSeq)

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

@@ -0,0 +1,57 @@
+/*
+ * 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.
+ */
+
+package za.co.absa.atum.server.api.database.runs.functions
+
+import doobie.Fragment
+import doobie.implicits.toSqlInterpolator
+import doobie.util.Read
+import play.api.libs.json.Json
+import za.co.absa.atum.model.dto.PartitioningDTO
+import za.co.absa.atum.server.api.database.PostgresDatabaseProvider
+import za.co.absa.atum.server.api.database.runs.Runs
+import za.co.absa.atum.server.model.PartitioningForDB
+import za.co.absa.fadb.DBSchema
+import za.co.absa.fadb.doobie.DoobieFunction.DoobieMultipleResultFunction
+import za.co.absa.fadb.doobie.DoobieEngine
+import zio.interop.catz.asyncInstance
+import zio.{Task, URLayer, ZIO, ZLayer}
+import za.co.absa.atum.server.api.database.DoobieImplicits.getMapWithOptionStringValues
+
+class GetPartitioningAdditionalData (implicit schema: DBSchema, dbEngine: DoobieEngine[Task])
+  extends DoobieMultipleResultFunction[PartitioningDTO, (String, Option[String]), Task]
+  {
+    import za.co.absa.atum.server.api.database.DoobieImplicits.Jsonb.jsonbPutUsingString
+    override def sql(values: PartitioningDTO)(implicit read: Read[(String, Option[String])]): Fragment = {
+    val partitioning: PartitioningForDB = PartitioningForDB.fromSeqPartitionDTO(values)
+    val partitioningJsonString = Json.toJson(partitioning).toString
+
+    sql"""SELECT ${Fragment.const(selectEntry)} FROM ${Fragment.const(functionName)}(
+                  ${
+                      partitioningJsonString
+                   }
+                ) ${Fragment.const(alias)};"""
+  }
+
+}
+
+object GetPartitioningAdditionalData {
+  val layer: URLayer[PostgresDatabaseProvider, GetPartitioningAdditionalData] = ZLayer {
+    for {
+      dbProvider <- ZIO.service[PostgresDatabaseProvider]
+    } yield new GetPartitioningAdditionalData()(Runs, dbProvider.dbEngine)
+  }
+}

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

@@ -0,0 +1,59 @@
+/*
+ * 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.
+ */
+
+package za.co.absa.atum.server.api.database.runs.functions
+
+import doobie.Fragment
+import doobie.implicits.toSqlInterpolator
+import doobie.util.Read
+import play.api.libs.json.Json
+import za.co.absa.atum.model.dto.{MeasureDTO, PartitioningDTO}
+import za.co.absa.atum.server.model.PartitioningForDB
+import za.co.absa.fadb.DBSchema
+import za.co.absa.fadb.doobie.DoobieEngine
+import za.co.absa.fadb.doobie.DoobieFunction.DoobieMultipleResultFunction
+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._
+import za.co.absa.atum.server.api.database.DoobieImplicits.Sequence.get
+
+class GetPartitioningMeasures (implicit schema: DBSchema, dbEngine: DoobieEngine[Task])
+  extends DoobieMultipleResultFunction[PartitioningDTO, MeasureDTO, Task]
+  {
+    import za.co.absa.atum.server.api.database.DoobieImplicits.Jsonb.jsonbPutUsingString
+
+    override def sql(values: PartitioningDTO)(implicit read: Read[MeasureDTO]): Fragment = {
+    val partitioning = PartitioningForDB.fromSeqPartitionDTO(values)
+    val partitioningJsonString = Json.toJson(partitioning).toString
+
+    sql"""SELECT ${Fragment.const(selectEntry)} FROM ${Fragment.const(functionName)}(
+                  ${
+                    partitioningJsonString
+                  }
+                ) ${Fragment.const(alias)};"""
+  }
+
+}
+
+object GetPartitioningMeasures {
+  val layer: URLayer[PostgresDatabaseProvider, GetPartitioningMeasures] = ZLayer {
+    for {
+      dbProvider <- ZIO.service[PostgresDatabaseProvider]
+    } yield new GetPartitioningMeasures()(Runs, dbProvider.dbEngine)
+  }
+}
+