Changes Made:

- Added the aggregatedTruncTotal Measure and the absAggregatedTruncTotal Measure.
- Added the tests for these Measures.
- Made amendments to the function to not include double casts.
- Changed the result from BigDecimal to LongValue.

Author: ABLL526

agent/src/main/scala/za/co/absa/atum/agent/model/Measure.scala

@@ -39,16 +39,6 @@ final case class UnknownMeasure(measureName: String, measuredColumns: Seq[String
 
 object AtumMeasure {
 
-//  val supportedMeasureNames: Seq[String] = Seq(
-//    RecordCount.measureName,
-//    DistinctRecordCount.measureName,
-//    SumOfValuesOfColumn.measureName,
-//    AbsSumOfValuesOfColumn.measureName,
-//    SumOfTruncatedValuesOfColumn.measureName,
-//    AbsSumOfTruncatedValuesOfColumn.measureName,
-//    SumOfHashesOfColumn.measureName
-//  )
-
   case class RecordCount private (measureName: String) extends AtumMeasure {
     private val columnExpression = count("*")
 
@@ -120,8 +110,7 @@ object AtumMeasure {
   }
 
   case class SumOfTruncatedValuesOfColumn private (measureName: String, measuredCol: String) extends AtumMeasure {
-    //Cast to LongType to remove decimal points then cast back to decimal to ensure compatibility
-    //private val columnAggFn: Column => Column = column => sum(column.cast(LongType).cast(DecimalType(38, 0)))
+
     private val columnAggFn: Column => Column = column => sum(when(column >= 0, floor(column)).otherwise(ceil(column)))
 
     override def function: MeasurementFunction = (ds: DataFrame) => {
@@ -131,16 +120,15 @@ object AtumMeasure {
     }
 
     override def measuredColumns: Seq[String] = Seq(measuredCol)
-    override val resultValueType: ResultValueType = ResultValueType.BigDecimalValue
+    override val resultValueType: ResultValueType = ResultValueType.LongValue
   }
   object SumOfTruncatedValuesOfColumn {
     private[agent] val measureName: String = "aggregatedTruncTotal"
     def apply(measuredCol: String): SumOfTruncatedValuesOfColumn = SumOfTruncatedValuesOfColumn(measureName, measuredCol)
   }
 
   case class AbsSumOfTruncatedValuesOfColumn private (measureName: String, measuredCol: String) extends AtumMeasure {
-    //Cast to LongType to remove decimal points then cast back to decimal to ensure compatibility
-    //private val columnAggFn: Column => Column = column => sum(abs(column.cast(LongType).cast(DecimalType(38, 0))))
+
     private val columnAggFn: Column => Column = column => sum(abs(when(column >= 0, floor(column)).otherwise(ceil(column))))
 
     override def function: MeasurementFunction = (ds: DataFrame) => {
@@ -150,7 +138,7 @@ object AtumMeasure {
     }
 
     override def measuredColumns: Seq[String] = Seq(measuredCol)
-    override val resultValueType: ResultValueType = ResultValueType.BigDecimalValue
+    override val resultValueType: ResultValueType = ResultValueType.LongValue
   }
   object AbsSumOfTruncatedValuesOfColumn {
     private[agent] val measureName: String = "absAggregatedTruncTotal"

agent/src/test/scala/za/co/absa/atum/agent/model/AtumMeasureUnitTests.scala

@@ -129,13 +129,13 @@ class AtumMeasureUnitTests extends AnyFlatSpec with Matchers with SparkTestBase
     assert(dfFullSalarySumResult.resultValue == "2987144")
     assert(dfFullSalarySumResult.resultValueType == ResultValueType.BigDecimalValue)
     assert(dfExtraPersonSalarySumTruncResult.resultValue == "2989144")
-    assert(dfExtraPersonSalarySumTruncResult.resultValueType == ResultValueType.BigDecimalValue)
+    assert(dfExtraPersonSalarySumTruncResult.resultValueType == ResultValueType.LongValue)
     assert(dfFullSalarySumTruncResult.resultValue == "2987144")
-    assert(dfFullSalarySumTruncResult.resultValueType == ResultValueType.BigDecimalValue)
+    assert(dfFullSalarySumTruncResult.resultValueType == ResultValueType.LongValue)
     assert(dfExtraPersonSalaryAbsSumTruncResult.resultValue == "2991144")
-    assert(dfExtraPersonSalaryAbsSumTruncResult.resultValueType == ResultValueType.BigDecimalValue)
+    assert(dfExtraPersonSalaryAbsSumTruncResult.resultValueType == ResultValueType.LongValue)
     assert(dfFullSalaryAbsSumTruncResult.resultValue == "2987144")
-    assert(dfFullSalaryAbsSumTruncResult.resultValueType == ResultValueType.BigDecimalValue)
+    assert(dfFullSalaryAbsSumTruncResult.resultValueType == ResultValueType.LongValue)
   }
 
   "AbsSumOfValuesOfColumn" should "return expected value" in {
@@ -229,7 +229,7 @@ class AtumMeasureUnitTests extends AnyFlatSpec with Matchers with SparkTestBase
     val result = distinctCount.function(df)
 
     assert(result.resultValue == "2")
-    assert(result.resultValueType == ResultValueType.BigDecimalValue)
+    assert(result.resultValueType == ResultValueType.LongValue)
   }
 
   "AbsSumTruncOfValuesOfColumn" should "return expected value" in {
@@ -244,6 +244,6 @@ class AtumMeasureUnitTests extends AnyFlatSpec with Matchers with SparkTestBase
     val result = distinctCount.function(df)
 
     assert(result.resultValue == "4")
-    assert(result.resultValueType == ResultValueType.BigDecimalValue)
+    assert(result.resultValueType == ResultValueType.LongValue)
   }
 }

agent/src/test/scala/za/co/absa/atum/agent/model/MeasureUnitTests.scala

@@ -127,13 +127,13 @@ class MeasureUnitTests extends AnyFlatSpec with Matchers with SparkTestBase { se
     assert(dfFullSalarySumResult.resultValue == "2987144")
     assert(dfFullSalarySumResult.resultValueType == ResultValueType.BigDecimalValue)
     assert(dfExtraPersonSalarySumTruncResult.resultValue == "2989144")
-    assert(dfExtraPersonSalarySumTruncResult.resultValueType == ResultValueType.BigDecimalValue)
+    assert(dfExtraPersonSalarySumTruncResult.resultValueType == ResultValueType.LongValue)
     assert(dfFullSalarySumTruncResult.resultValue == "2987144")
-    assert(dfFullSalarySumTruncResult.resultValueType == ResultValueType.BigDecimalValue)
+    assert(dfFullSalarySumTruncResult.resultValueType == ResultValueType.LongValue)
     assert(dfExtraPersonSalaryAbsSumTruncResult.resultValue == "2991144")
-    assert(dfExtraPersonSalaryAbsSumTruncResult.resultValueType == ResultValueType.BigDecimalValue)
+    assert(dfExtraPersonSalaryAbsSumTruncResult.resultValueType == ResultValueType.LongValue)
     assert(dfFullSalaryAbsSumTruncResult.resultValue == "2987144")
-    assert(dfFullSalaryAbsSumTruncResult.resultValueType == ResultValueType.BigDecimalValue)
+    assert(dfFullSalaryAbsSumTruncResult.resultValueType == ResultValueType.LongValue)
   }
 
 }