#26 add create checkpoint function (#79)

* #26 add checkpoint creation function

* #26: finishing bear minimum of agent side

* #26: post-review improvements and refactoring

* #26: extra validation, putting emphasis on types, some generics, more unit tests, refactoring & simplification

* #26: functionName -> measureName, to be consistent

* #26: post-review changes

* #26: fixing unit test

* #26: removing redundant internal Agent's Checkpoint model and some refactoring

* #26: removing onlyForNumeric as it's not currently used and we might do this differently

* #26: post-review changes

---------

Co-authored-by: Ladislav Sulak <laco.sulak@gmail.com>

Author: cerveada

LICENSE.md

@@ -199,4 +199,4 @@
    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.
+   limitations under the License.

agent/src/main/scala/za/co/absa/atum/agent/AtumAgent.scala

@@ -18,11 +18,10 @@ package za.co.absa.atum.agent
 import com.typesafe.config.{Config, ConfigFactory}
 import za.co.absa.atum.agent.AtumContext.AtumPartitions
 import za.co.absa.atum.agent.dispatcher.{ConsoleDispatcher, HttpDispatcher}
-import za.co.absa.atum.agent.model.Checkpoint
 import za.co.absa.atum.model.dto.{CheckpointDTO, PartitioningDTO}
 
 /**
- * Place holder for the agent that communicate with the API.
+ * Entity that communicate with the API, primarily focused on spawning Atum Context(s).
  */
 class AtumAgent private[agent] () {
 
@@ -36,23 +35,16 @@ class AtumAgent private[agent] () {
 
   /**
    * Sends `CheckpointDTO` to the AtumService API
-   * @param checkpoint
-   */
-  def saveCheckpoint(checkpoint: CheckpointDTO): Unit = {
-    dispatcher.saveCheckpoint(checkpoint)
-  }
-
-  /**
-   * Sends `Checkpoint` to the AtumService API
    *
-   * @param checkpoint
+   * @param checkpoint Already initialized Checkpoint object to store
    */
-  def saveCheckpoint(checkpoint: Checkpoint): Unit = {
-    dispatcher.saveCheckpoint(checkpoint.toCheckpointDTO)
+  private [agent] def saveCheckpoint(checkpoint: CheckpointDTO): Unit = {
+    dispatcher.saveCheckpoint(checkpoint)
   }
 
   /**
    *  Provides an AtumContext given a `AtumPartitions` instance. Retrieves the data from AtumService API.
+   *
    *  @param atumPartitions
    *  @return
    */

agent/src/main/scala/za/co/absa/atum/agent/AtumContext.scala

@@ -18,10 +18,12 @@ package za.co.absa.atum.agent
 
 import org.apache.spark.sql.DataFrame
 import za.co.absa.atum.agent.AtumContext.AtumPartitions
-import za.co.absa.atum.agent.model.{Checkpoint, Measure, Measurement, MeasuresMapper}
-import za.co.absa.atum.model.dto.{AtumContextDTO, PartitionDTO}
+import za.co.absa.atum.agent.model.Measurement.MeasurementByAtum
+import za.co.absa.atum.agent.model._
+import za.co.absa.atum.model.dto._
 
 import java.time.OffsetDateTime
+import java.util.UUID
 import scala.collection.immutable.ListMap
 
 /**
@@ -43,24 +45,48 @@ class AtumContext private[agent] (
     agent.getOrCreateAtumSubContext(atumPartitions ++ subPartitions)(this)
   }
 
-  def createCheckpoint(checkpointName: String, author: String, dataToMeasure: DataFrame) = {
-    ??? // TODO #26
+  private def takeMeasurements(df: DataFrame): Set[MeasurementByAtum] = {
+    measures.map { m =>
+      val measurementResult = m.function(df)
+      MeasurementByAtum(m, measurementResult.result, measurementResult.resultType)
+    }
+  }
+
+  def createCheckpoint(checkpointName: String, author: String, dataToMeasure: DataFrame): AtumContext = {
+    val startTime = OffsetDateTime.now()
+    val measurements = takeMeasurements(dataToMeasure)
+    val endTime = OffsetDateTime.now()
+
+    val checkpointDTO = CheckpointDTO(
+      id = UUID.randomUUID(),
+      name = checkpointName,
+      author = author,
+      measuredByAtumAgent = true,
+      partitioning = AtumPartitions.toSeqPartitionDTO(this.atumPartitions),
+      processStartTime = startTime,
+      processEndTime = Some(endTime),
+      measurements = measurements.map(MeasurementBuilder.buildMeasurementDTO).toSeq
+    )
+
+    agent.saveCheckpoint(checkpointDTO)
+    this
   }
 
-  def createCheckpointOnProvidedData(
-                                      checkpointName: String,
-                                      author: String,
-                                      measurements: Seq[Measurement]
-                                    ): Checkpoint = {
+  def createCheckpointOnProvidedData(checkpointName: String, author: String, measurements: Seq[Measurement]): AtumContext = {
     val offsetDateTimeNow = OffsetDateTime.now()
-    Checkpoint(
+
+    val checkpointDTO = CheckpointDTO(
+      id = UUID.randomUUID(),
       name = checkpointName,
       author = author,
-      atumPartitions = this.atumPartitions,
+      partitioning = AtumPartitions.toSeqPartitionDTO(this.atumPartitions),
       processStartTime = offsetDateTimeNow,
       processEndTime = Some(offsetDateTimeNow),
-      measurements = measurements
+      measurements = measurements.map(MeasurementBuilder.buildMeasurementDTO)
     )
+
+    agent.saveCheckpoint(checkpointDTO)
+    this
   }
 
   def addAdditionalData(key: String, value: String): Unit = {
@@ -121,21 +147,22 @@ object AtumContext {
     new AtumContext(
       AtumPartitions.fromPartitioning(atumContextDTO.partitioning),
       agent,
-      MeasuresMapper.mapToMeasures(atumContextDTO.measures),
+      MeasuresBuilder.mapToMeasures(atumContextDTO.measures),
       atumContextDTO.additionalData.additionalData
     )
   }
 
   implicit class DatasetWrapper(df: DataFrame) {
 
     /**
-     *  Set a point in the pipeline to execute calculation.
+     *  Set a point in the pipeline to execute calculation and store it.
      *  @param checkpointName The key assigned to this checkpoint
+     *  @param author Author of the checkpoint
      *  @param atumContext Contains the calculations to be done and publish the result
      *  @return
      */
     def createCheckpoint(checkpointName: String, author: String)(implicit atumContext: AtumContext): DataFrame = {
-      // todo: implement checkpoint creation
+      atumContext.createCheckpoint(checkpointName, author, df)
       df
     }
 

agent/src/main/scala/za/co/absa/atum/agent/core/MeasurementProcessor.scala

@@ -18,6 +18,7 @@ package za.co.absa.atum.agent.core
 
 import org.apache.spark.sql.DataFrame
 import za.co.absa.atum.agent.core.MeasurementProcessor.MeasurementFunction
+import za.co.absa.atum.model.dto.MeasureResultDTO.ResultValueType
 
 trait MeasurementProcessor {
 
@@ -26,6 +27,13 @@ trait MeasurementProcessor {
 }
 
 object MeasurementProcessor {
-  type MeasurementFunction = DataFrame => String
+  /**
+   * The raw result of measurement is always gonna be string, because we want to avoid some floating point issues
+   * (overflows, consistent representation of numbers - whether they are coming from Java or Scala world, and more),
+   * but the actual type is stored alongside the computation because we don't want to lost this information.
+   */
+  final case class ResultOfMeasurement(result: String, resultType: ResultValueType.ResultValueType)
+
+  type MeasurementFunction = DataFrame => ResultOfMeasurement
 
 }

agent/src/main/scala/za/co/absa/atum/agent/dispatcher/ConsoleDispatcher.scala

@@ -33,4 +33,5 @@ class ConsoleDispatcher extends Dispatcher with Logging {
   override def saveCheckpoint(checkpoint: CheckpointDTO): Unit = {
     println(s"Saving checkpoint to server. $checkpoint")
   }
+
 }

agent/src/main/scala/za/co/absa/atum/agent/dispatcher/HttpDispatcher.scala

@@ -45,4 +45,5 @@ class HttpDispatcher(config: Config) extends Dispatcher with Logging {
       .post(serverUri)
       .send(backend)
   }
+
 }

agent/src/main/scala/za/co/absa/atum/agent/exception/UnsupportedMeasureException.scala agent/src/main/scala/za/co/absa/atum/agent/exception/AtumAgentException.scala

@@ -16,4 +16,7 @@
 
 package za.co.absa.atum.agent.exception
 
-case class UnsupportedMeasureException(msg: String) extends Exception(msg)
+sealed abstract class AtumAgentException extends Exception
+
+case class MeasurementProvidedException(msg: String) extends AtumAgentException
+case class MeasureException(msg: String) extends AtumAgentException

agent/src/main/scala/za/co/absa/atum/agent/exception/UnsupportedMeasureResultType.scala

@@ -20,7 +20,8 @@ import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.{DecimalType, LongType, StringType}
 import org.apache.spark.sql.{Column, DataFrame}
 import za.co.absa.atum.agent.core.MeasurementProcessor
-import za.co.absa.atum.agent.core.MeasurementProcessor.MeasurementFunction
+import za.co.absa.atum.agent.core.MeasurementProcessor.{MeasurementFunction, ResultOfMeasurement}
+import za.co.absa.atum.model.dto.MeasureResultDTO.ResultValueType
 import za.co.absa.spark.commons.implicits.StructTypeImplicits.StructTypeEnhancements
 
 /**
@@ -32,86 +33,109 @@ sealed trait Measure extends MeasurementProcessor with MeasureType {
 
 trait MeasureType {
   val measureName: String
-  val onlyForNumeric: Boolean
+  val resultValueType: ResultValueType.ResultValueType
 }
 
 object Measure {
 
   private val valueColumnName: String = "value"
 
   val supportedMeasures: Seq[MeasureType] = Seq(
-    RecordCount, DistinctRecordCount, SumOfValuesOfColumn, AbsSumOfValuesOfColumn, SumOfHashesOfColumn
+    RecordCount,
+    DistinctRecordCount,
+    SumOfValuesOfColumn,
+    AbsSumOfValuesOfColumn,
+    SumOfHashesOfColumn
   )
   val supportedMeasureNames: Seq[String] = supportedMeasures.map(_.measureName)
 
-  case class RecordCount private (controlCol: String, measureName: String, onlyForNumeric: Boolean) extends Measure {
+  case class RecordCount private (
+    controlCol: String,
+    measureName: String,
+    resultValueType: ResultValueType.ResultValueType
+  ) extends Measure {
 
     override def function: MeasurementFunction =
-      (ds: DataFrame) => ds.select(col(controlCol)).count().toString
+      (ds: DataFrame) => {
+        val resultValue = ds.select(col(controlCol)).count().toString
+        ResultOfMeasurement(resultValue, resultValueType)
+      }
   }
   object RecordCount extends MeasureType {
-    def apply(controlCol: String): RecordCount = {
-      RecordCount(controlCol, measureName, onlyForNumeric)
-    }
+    def apply(controlCol: String): RecordCount = RecordCount(controlCol, measureName, resultValueType)
 
     override val measureName: String = "count"
-    override val onlyForNumeric: Boolean = false
+    override val resultValueType: ResultValueType.ResultValueType = ResultValueType.Long
   }
 
-  case class DistinctRecordCount private (controlCol: String, measureName: String, onlyForNumeric: Boolean)
-      extends Measure {
+  case class DistinctRecordCount private (
+    controlCol: String,
+    measureName: String,
+    resultValueType: ResultValueType.ResultValueType
+  ) extends Measure {
 
     override def function: MeasurementFunction =
-      (ds: DataFrame) => ds.select(col(controlCol)).distinct().count().toString
+      (ds: DataFrame) => {
+        val resultValue = ds.select(col(controlCol)).distinct().count().toString
+        ResultOfMeasurement(resultValue, resultValueType)
+      }
   }
-
   object DistinctRecordCount extends MeasureType {
     def apply(controlCol: String): DistinctRecordCount = {
-      DistinctRecordCount(controlCol, measureName, onlyForNumeric)
+      DistinctRecordCount(controlCol, measureName, resultValueType)
     }
 
     override val measureName: String = "distinctCount"
-    override val onlyForNumeric: Boolean = false
+    override val resultValueType: ResultValueType.ResultValueType = ResultValueType.Long
   }
 
-  case class SumOfValuesOfColumn private (controlCol: String, measureName: String, onlyForNumeric: Boolean)
-      extends Measure {
+  case class SumOfValuesOfColumn private (
+    controlCol: String,
+    measureName: String,
+    resultValueType: ResultValueType.ResultValueType
+  ) extends Measure {
 
     override def function: MeasurementFunction = (ds: DataFrame) => {
       val aggCol = sum(col(valueColumnName))
-      aggregateColumn(ds, controlCol, aggCol)
+      val resultValue = aggregateColumn(ds, controlCol, aggCol)
+      ResultOfMeasurement(resultValue, resultValueType)
     }
   }
-
   object SumOfValuesOfColumn extends MeasureType {
     def apply(controlCol: String): SumOfValuesOfColumn = {
-      SumOfValuesOfColumn(controlCol, measureName, onlyForNumeric)
+      SumOfValuesOfColumn(controlCol, measureName, resultValueType)
     }
 
     override val measureName: String = "aggregatedTotal"
-    override val onlyForNumeric: Boolean = true
+    override val resultValueType: ResultValueType.ResultValueType = ResultValueType.BigDecimal
   }
 
-  case class AbsSumOfValuesOfColumn private (controlCol: String, measureName: String, onlyForNumeric: Boolean)
-      extends Measure {
+  case class AbsSumOfValuesOfColumn private (
+    controlCol: String,
+    measureName: String,
+    resultValueType: ResultValueType.ResultValueType
+  ) extends Measure {
 
     override def function: MeasurementFunction = (ds: DataFrame) => {
       val aggCol = sum(abs(col(valueColumnName)))
-      aggregateColumn(ds, controlCol, aggCol)
+      val resultValue = aggregateColumn(ds, controlCol, aggCol)
+      ResultOfMeasurement(resultValue, resultValueType)
     }
   }
-
   object AbsSumOfValuesOfColumn extends MeasureType {
     def apply(controlCol: String): AbsSumOfValuesOfColumn = {
-      AbsSumOfValuesOfColumn(controlCol, measureName, onlyForNumeric)
+      AbsSumOfValuesOfColumn(controlCol, measureName, resultValueType)
     }
 
     override val measureName: String = "absAggregatedTotal"
-    override val onlyForNumeric: Boolean = true
+    override val resultValueType: ResultValueType.ResultValueType = ResultValueType.Double
   }
 
-  case class SumOfHashesOfColumn private (controlCol: String, measureName: String, onlyForNumeric: Boolean)
-      extends Measure {
+  case class SumOfHashesOfColumn private (
+    controlCol: String,
+    measureName: String,
+    resultValueType: ResultValueType.ResultValueType
+  ) extends Measure {
 
     override def function: MeasurementFunction = (ds: DataFrame) => {
 
@@ -120,17 +144,17 @@ object Measure {
         .withColumn(aggregatedColumnName, crc32(col(controlCol).cast("String")))
         .agg(sum(col(aggregatedColumnName)))
         .collect()(0)(0)
-      if (value == null) "" else value.toString
+      val resultValue = if (value == null) "" else value.toString
+      ResultOfMeasurement(resultValue, ResultValueType.String)
     }
   }
-
   object SumOfHashesOfColumn extends MeasureType {
     def apply(controlCol: String): SumOfHashesOfColumn = {
-      SumOfHashesOfColumn(controlCol, measureName, onlyForNumeric)
+      SumOfHashesOfColumn(controlCol, measureName, resultValueType)
     }
 
     override val measureName: String = "hashCrc32"
-    override val onlyForNumeric: Boolean = false
+    override val resultValueType: ResultValueType.ResultValueType = ResultValueType.String
   }
 
   private def aggregateColumn(

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

@@ -16,4 +16,65 @@
 
 package za.co.absa.atum.agent.model
 
-case class Measurement(measure: Measure, result: Any)
+import za.co.absa.atum.agent.exception.MeasurementProvidedException
+import za.co.absa.atum.model.dto.MeasureResultDTO.ResultValueType
+
+trait Measurement {
+  val measure: Measure
+  val resultValue: Any
+  val resultType: ResultValueType.ResultValueType
+}
+
+object Measurement {
+
+  /**
+   * When the application/user of Atum Agent provides actual results by himself, the type is precise and we don't need
+   * to do any adjustments.
+   */
+  case class MeasurementProvided[T](measure: Measure, resultValue: T, resultType: ResultValueType.ResultValueType)
+    extends Measurement
+
+  object MeasurementProvided {
+
+    private def handleSpecificType[T](
+                               measure: Measure, resultValue: T, requiredType: ResultValueType.ResultValueType
+                             ): MeasurementProvided[T] = {
+
+      val actualType = measure.resultValueType
+      if (actualType != requiredType)
+        throw MeasurementProvidedException(
+          s"Type of a given provided measurement result and type that a given measure supports are not compatible! " +
+            s"Got $actualType but should be $requiredType"
+        )
+      MeasurementProvided[T](measure, resultValue, requiredType)
+    }
+
+    def apply[T](measure: Measure, resultValue: T): Measurement = {
+      resultValue match {
+        case l: Long =>
+          handleSpecificType[Long](measure, l, ResultValueType.Long)
+        case d: Double =>
+          handleSpecificType[Double](measure, d, ResultValueType.Double)
+        case bd: BigDecimal =>
+          handleSpecificType[BigDecimal](measure, bd, ResultValueType.BigDecimal)
+        case s: String =>
+          handleSpecificType[String](measure, s, ResultValueType.String)
+
+        case unsupportedType =>
+          val className = unsupportedType.getClass.getSimpleName
+          throw MeasurementProvidedException(
+            s"Unsupported type of measurement for measure ${measure.measureName}: $className " +
+              s"for provided result: $resultValue"
+          )
+      }
+    }
+  }
+
+  /**
+   * When the Atum Agent itself performs the measurements, using Spark, then in some cases some adjustments are
+   * needed - thus we are converting the results to strings always - but we need to keep the information about
+   * the actual type as well.
+   */
+  case class MeasurementByAtum(measure: Measure, resultValue: String, resultType: ResultValueType.ResultValueType)
+    extends Measurement
+}

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

@@ -16,55 +16,18 @@
 
 package za.co.absa.atum.agent.model
 
-import za.co.absa.atum.agent.exception.UnsupportedMeasureResultType
 import za.co.absa.atum.model.dto.{MeasureDTO, MeasureResultDTO, MeasurementDTO}
-import za.co.absa.atum.model.dto.MeasureResultDTO.{ResultValueType, TypedValue}
+import za.co.absa.atum.model.dto.MeasureResultDTO.TypedValue
 
 private [agent] object MeasurementBuilder {
 
-  def buildMeasurementDTO(measurement: Measurement): MeasurementDTO = {
+  private [agent] def buildMeasurementDTO(measurement: Measurement): MeasurementDTO = {
     val measureName = measurement.measure.measureName
-    measurement.result match {
-      case l: Long =>
-        buildLongMeasurement(measureName, Seq(measurement.measure.controlCol), l)
-      case d: Double =>
-        buildDoubleMeasureResult(measureName, Seq(measurement.measure.controlCol), d)
-      case bd: BigDecimal =>
-        buildBigDecimalMeasureResult(measureName, Seq(measurement.measure.controlCol), bd)
-      case s: String =>
-        buildStringMeasureResult(measureName, Seq(measurement.measure.controlCol), s)
-      case unsupportedType  =>
-        val className = unsupportedType.getClass.getSimpleName
-        throw UnsupportedMeasureResultType(s"Unsupported type of measure $measureName: $className")
-    }
-  }
-
-  private def buildLongMeasurement(functionName: String, controlCols: Seq[String], resultValue: Long): MeasurementDTO = {
-    MeasurementDTO(
-      MeasureDTO(functionName, controlCols),
-      MeasureResultDTO(TypedValue(resultValue.toString, ResultValueType.Long))
-    )
-  }
+    val controlCols = Seq(measurement.measure.controlCol)
+    val measureDTO = MeasureDTO(measureName, controlCols)
 
-  private def buildDoubleMeasureResult(functionName: String, controlCols: Seq[String], resultValue: Double): MeasurementDTO = {
-    MeasurementDTO(
-      MeasureDTO(functionName, controlCols),
-      MeasureResultDTO(TypedValue(resultValue.toString, ResultValueType.Double))
-    )
-  }
+    val measureResultDTO = MeasureResultDTO(TypedValue(measurement.resultValue.toString, measurement.resultType))
 
-  private def buildBigDecimalMeasureResult(functionName: String, controlCols: Seq[String], resultValue: BigDecimal): MeasurementDTO = {
-    MeasurementDTO(
-      MeasureDTO(functionName, controlCols),
-      MeasureResultDTO(TypedValue(resultValue.toString, ResultValueType.BigDecimal))
-    )
+    MeasurementDTO(measureDTO, measureResultDTO)
   }
-
-  private def buildStringMeasureResult(functionName: String, controlCols: Seq[String], resultValue: String): MeasurementDTO = {
-    MeasurementDTO(
-      MeasureDTO(functionName, controlCols),
-      MeasureResultDTO(TypedValue(resultValue, ResultValueType.String))
-    )
-  }
-
 }

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

@@ -16,27 +16,27 @@
 
 package za.co.absa.atum.agent.model
 
-import za.co.absa.atum.agent.exception.UnsupportedMeasureException
+import za.co.absa.atum.agent.exception.MeasureException
 import za.co.absa.atum.agent.model.Measure._
 import za.co.absa.atum.model.dto
 
-private [agent] object MeasuresMapper {
+private [agent] object MeasuresBuilder {
 
-  def mapToMeasures(measures: Set[dto.MeasureDTO]): Set[za.co.absa.atum.agent.model.Measure] = {
+  private [agent] def mapToMeasures(measures: Set[dto.MeasureDTO]): Set[za.co.absa.atum.agent.model.Measure] = {
     measures.map(createMeasure)
   }
 
   private def createMeasure(measure: dto.MeasureDTO): za.co.absa.atum.agent.model.Measure = {
     val controlColumn = measure.controlColumns.head
 
-    measure.functionName match {
+    measure.measureName match {
       case RecordCount.measureName            => RecordCount(controlColumn)
       case DistinctRecordCount.measureName    => DistinctRecordCount(controlColumn)
       case SumOfValuesOfColumn.measureName    => SumOfValuesOfColumn(controlColumn)
       case AbsSumOfValuesOfColumn.measureName => AbsSumOfValuesOfColumn(controlColumn)
       case SumOfHashesOfColumn.measureName    => SumOfHashesOfColumn(controlColumn)
       case unsupportedMeasure =>
-        throw UnsupportedMeasureException(
+        throw MeasureException(
           s"Measure not supported: $unsupportedMeasure. Supported measures are: ${Measure.supportedMeasureNames}"
         )
     }

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

@@ -16,11 +16,19 @@
 
 package za.co.absa.atum.agent
 
+import org.apache.spark.sql.types.DoubleType
+import org.apache.spark.sql.{Row, SparkSession}
+import org.mockito.Mockito.{mock, times, verify}
+import org.mockito.ArgumentCaptor
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import za.co.absa.atum.agent.AtumContext.AtumPartitions
 import za.co.absa.atum.agent.model.Measure.{RecordCount, SumOfValuesOfColumn}
-import za.co.absa.atum.agent.model.Measurement
+import za.co.absa.atum.agent.model.MeasurementBuilder
+import za.co.absa.atum.agent.model.Measurement.MeasurementProvided
+import org.apache.spark.sql.types.{StringType, StructField, StructType}
+import za.co.absa.atum.model.dto.CheckpointDTO
+import za.co.absa.atum.model.dto.MeasureResultDTO.ResultValueType
 
 class AtumContextTest extends AnyFlatSpec with Matchers {
 
@@ -46,7 +54,6 @@ class AtumContextTest extends AnyFlatSpec with Matchers {
       )
 
     assert(atumContextWithTwoDistinctRecordCount.currentMeasures.size == 3)
-
   }
 
   "withMeasureRemoved" should "remove a measure if exists" in {
@@ -64,25 +71,112 @@ class AtumContextTest extends AnyFlatSpec with Matchers {
     assert(atumContextRemoved.currentMeasures.head == RecordCount("other"))
   }
 
+  "createCheckpoint" should "take measurements and create a Checkpoint" in {
+    val mockAgent = mock(classOf[AtumAgent])
+    val atumPartitions = AtumPartitions("foo2" -> "bar")
+
+    val atumContext = new AtumContext(atumPartitions, mockAgent)
+      .addMeasure(RecordCount("letter"))
+
+    val spark = SparkSession.builder
+      .master("local")
+      .config("spark.driver.host", "localhost")
+      .config("spark.ui.enabled", "false")
+      .getOrCreate()
+
+    import spark.implicits._
+
+    val rdd = spark.sparkContext.parallelize(Seq("A", "B", "C"))
+    val df = rdd.toDF("letter")
+
+    atumContext.createCheckpoint("testCheckpoint", "Hans", df)
+
+    val argument = ArgumentCaptor.forClass(classOf[CheckpointDTO])
+    verify(mockAgent).saveCheckpoint(argument.capture())
+
+    assert(argument.getValue.name == "testCheckpoint")
+    assert(argument.getValue.author == "Hans")
+    assert(argument.getValue.partitioning == AtumPartitions.toSeqPartitionDTO(atumPartitions))
+    assert(argument.getValue.measurements.head.result.mainValue.value == "3")
+    assert(argument.getValue.measurements.head.result.mainValue.valueType == ResultValueType.Long)
+  }
+
   "createCheckpointOnProvidedData" should "create a Checkpoint on provided data" in {
-    val atumAgent = new AtumAgent
+    val mockAgent = mock(classOf[AtumAgent])
     val atumPartitions = AtumPartitions("key" -> "value")
-    val atumContext = atumAgent.getOrCreateAtumContext(atumPartitions)
+    val atumContext: AtumContext = new AtumContext(atumPartitions, mockAgent)
 
-    val measurements = Seq(Measurement(RecordCount("col"), "1"), Measurement(SumOfValuesOfColumn("col"), 1))
+    val measurements = Seq(
+      MeasurementProvided(RecordCount("col"), 1L),
+      MeasurementProvided(SumOfValuesOfColumn("col"), BigDecimal(1))
+    )
 
-    val checkpoint = atumContext.createCheckpointOnProvidedData(
+    atumContext.createCheckpointOnProvidedData(
       checkpointName = "name",
       author = "author",
       measurements = measurements
     )
 
-    assert(checkpoint.name == "name")
-    assert(checkpoint.author == "author")
-    assert(!checkpoint.measuredByAtumAgent)
-    assert(checkpoint.atumPartitions == atumPartitions)
-    assert(checkpoint.processStartTime == checkpoint.processEndTime.get)
-    assert(checkpoint.measurements == measurements)
+    val argument = ArgumentCaptor.forClass(classOf[CheckpointDTO])
+    verify(mockAgent).saveCheckpoint(argument.capture())
+
+    assert(argument.getValue.name == "name")
+    assert(argument.getValue.author == "author")
+    assert(!argument.getValue.measuredByAtumAgent)
+    assert(argument.getValue.partitioning == AtumPartitions.toSeqPartitionDTO(atumPartitions))
+    assert(argument.getValue.processStartTime == argument.getValue.processEndTime.get)
+    assert(argument.getValue.measurements == measurements.map(MeasurementBuilder.buildMeasurementDTO))
+  }
+
+  "createCheckpoint" should "take measurements and create a Checkpoint, multiple measure changes" in {
+    val mockAgent = mock(classOf[AtumAgent])
+    val atumPartitions = AtumPartitions("foo2" -> "bar")
+    implicit val atumContext: AtumContext = new AtumContext(atumPartitions, mockAgent)
+      .addMeasure(RecordCount("notImportantColumn"))
+
+    val spark = SparkSession.builder
+      .master("local")
+      .config("spark.driver.host", "localhost")
+      .config("spark.ui.enabled", "false")
+      .getOrCreate()
+
+    val rdd = spark.sparkContext.parallelize(
+      Seq(
+        Row("A", 8.0),
+        Row("B", 2.9),
+        Row("C", 9.1),
+        Row("D", 2.5)
+      )
+    )
+    val schema = new StructType()
+      .add(StructField("notImportantColumn", StringType))
+      .add(StructField("columnForSum", DoubleType))
+
+    import AtumContext._
+
+    val df = spark.createDataFrame(rdd, schema)
+      .createCheckpoint("checkPointNameCount", "authorOfCount")
+
+    val argumentFirst = ArgumentCaptor.forClass(classOf[CheckpointDTO])
+    verify(mockAgent, times(1)).saveCheckpoint(argumentFirst.capture())
+
+    assert(argumentFirst.getValue.name == "checkPointNameCount")
+    assert(argumentFirst.getValue.author == "authorOfCount")
+    assert(argumentFirst.getValue.partitioning == AtumPartitions.toSeqPartitionDTO(atumPartitions))
+    assert(argumentFirst.getValue.measurements.head.result.mainValue.value == "4")
+    assert(argumentFirst.getValue.measurements.head.result.mainValue.valueType == ResultValueType.Long)
+
+    atumContext.addMeasure(SumOfValuesOfColumn("columnForSum"))
+    df.createCheckpoint("checkPointNameSum", "authorOfSum")
+
+    val argumentSecond = ArgumentCaptor.forClass(classOf[CheckpointDTO])
+    verify(mockAgent, times(2)).saveCheckpoint(argumentSecond.capture())
+
+    assert(argumentSecond.getValue.name == "checkPointNameSum")
+    assert(argumentSecond.getValue.author == "authorOfSum")
+    assert(argumentSecond.getValue.partitioning == AtumPartitions.toSeqPartitionDTO(atumPartitions))
+    assert(argumentSecond.getValue.measurements.tail.head.result.mainValue.value == "22.5")
+    assert(argumentSecond.getValue.measurements.tail.head.result.mainValue.valueType == ResultValueType.BigDecimal)
   }
 
   "addAdditionalData" should "add key/value pair to map for additional data" in {

agent/src/main/scala/za/co/absa/atum/agent/model/MeasuresMapper.scala agent/src/main/scala/za/co/absa/atum/agent/model/MeasuresBuilder.scala

@@ -21,6 +21,7 @@ import org.scalatest.matchers.should.Matchers
 import za.co.absa.atum.agent.AtumAgent
 import za.co.absa.atum.agent.AtumContext.{AtumPartitions, DatasetWrapper}
 import za.co.absa.atum.agent.model.Measure._
+import za.co.absa.atum.model.dto.MeasureResultDTO.ResultValueType
 import za.co.absa.spark.commons.test.SparkTestBase
 
 class MeasureTest extends AnyFlatSpec with Matchers with SparkTestBase { self =>
@@ -36,11 +37,14 @@ class MeasureTest extends AnyFlatSpec with Matchers with SparkTestBase { self =>
     val sumOfHashes: Measure = SumOfHashesOfColumn(controlCol = "id")
 
     // AtumContext contains `Measurement`
-    val atumContextInstanceWithRecordCount = AtumAgent.getOrCreateAtumContext(AtumPartitions("foo"->"bar"))
+    val atumContextInstanceWithRecordCount = AtumAgent
+      .getOrCreateAtumContext(AtumPartitions("foo"->"bar"))
       .addMeasure(measureIds)
-    val atumContextWithSalaryAbsMeasure = atumContextInstanceWithRecordCount.subPartitionContext(AtumPartitions("sub"->"partition"))
+    val atumContextWithSalaryAbsMeasure = atumContextInstanceWithRecordCount
+      .subPartitionContext(AtumPartitions("sub"->"partition"))
       .addMeasure(salaryAbsSum)
-    val atumContextWithNameHashSum = atumContextInstanceWithRecordCount.subPartitionContext(AtumPartitions("another"->"partition"))
+    val atumContextWithNameHashSum = atumContextInstanceWithRecordCount
+      .subPartitionContext(AtumPartitions("another"->"partition"))
       .addMeasure(sumOfHashes)
 
     // Pipeline
@@ -56,8 +60,7 @@ class MeasureTest extends AnyFlatSpec with Matchers with SparkTestBase { self =>
       .option("header", "true")
       .load("agent/src/test/resources/random-dataset/persons-enriched.csv")
       .createCheckpoint("name3", "author")(
-        atumContextWithSalaryAbsMeasure
-          .removeMeasure(salaryAbsSum)
+        atumContextWithSalaryAbsMeasure.removeMeasure(salaryAbsSum)
       )
 
     val dfFull = dfPersons
@@ -80,14 +83,26 @@ class MeasureTest extends AnyFlatSpec with Matchers with SparkTestBase { self =>
         .removeMeasure(salaryAbsSum)
     )
 
-    // Assertions
-    assert(measureIds.function(dfPersons) == "1000")
-    assert(measureIds.function(dfFull) == "1000")
-    assert(salaryAbsSum.function(dfFull) == "2987144")
-    assert(sumOfHashes.function(dfFull) == "2044144307532")
-    assert(salarySum.function(dfExtraPerson) == "2986144")
-    assert(salarySum.function(dfFull) == "2987144")
+    val dfPersonCntResult             = measureIds.function(dfPersons)
+    val dfFullCntResult               = measureIds.function(dfFull)
+    val dfFullSalaryAbsSumResult      = salaryAbsSum.function(dfFull)
+    val dfFullHashResult              = sumOfHashes.function(dfFull)
+    val dfExtraPersonSalarySumResult  = salarySum.function(dfExtraPerson)
+    val dfFullSalarySumResult         = salarySum.function(dfFull)
 
+    // Assertions
+    assert(dfPersonCntResult.result == "1000")
+    assert(dfPersonCntResult.resultType == ResultValueType.Long)
+    assert(dfFullCntResult.result == "1000")
+    assert(dfFullCntResult.resultType == ResultValueType.Long)
+    assert(dfFullSalaryAbsSumResult.result == "2987144")
+    assert(dfFullSalaryAbsSumResult.resultType == ResultValueType.Double)
+    assert(dfFullHashResult.result == "2044144307532")
+    assert(dfFullHashResult.resultType == ResultValueType.String)
+    assert(dfExtraPersonSalarySumResult.result == "2986144")
+    assert(dfExtraPersonSalarySumResult.resultType == ResultValueType.BigDecimal)
+    assert(dfFullSalarySumResult.result == "2987144")
+    assert(dfFullSalarySumResult.resultType == ResultValueType.BigDecimal)
   }
 
 }

agent/src/test/scala/za/co/absa/atum/agent/AtumContextTest.scala

@@ -17,63 +17,62 @@
 package za.co.absa.atum.agent.model
 
 import org.scalatest.flatspec.AnyFlatSpec
-import za.co.absa.atum.agent.exception.UnsupportedMeasureResultType
 import za.co.absa.atum.agent.model.Measure.SumOfValuesOfColumn
+import za.co.absa.atum.agent.model.Measurement.{MeasurementProvided, MeasurementByAtum}
 import za.co.absa.atum.model.dto.{MeasureDTO, MeasureResultDTO}
 import za.co.absa.atum.model.dto.MeasureResultDTO.{ResultValueType, TypedValue}
 
 class MeasurementBuilderTest extends AnyFlatSpec {
 
-  "buildMeasurementDTO" should "build expected MeasurementDTO for Long type of result value" in {
+  "buildMeasurementDTO" should "build MeasurementDTO for Long type of result value when Measurement provided" in {
     val measure = SumOfValuesOfColumn("col")
-    val measurement = Measurement(measure, 1L)
+    val measurement = MeasurementProvided(measure, BigDecimal(1))
     val measurementDTO = MeasurementBuilder.buildMeasurementDTO(measurement)
 
     val expectedMeasureDTO = MeasureDTO("aggregatedTotal", Seq("col"))
 
     val expectedMeasureResultDTO = MeasureResultDTO(
-      TypedValue("1", ResultValueType.Long)
+      TypedValue("1", ResultValueType.BigDecimal)
     )
 
     assert(measurementDTO.measure == expectedMeasureDTO)
     assert(measurementDTO.result == expectedMeasureResultDTO)
   }
 
-  "buildMeasurementDTO" should "build MeasurementDTO with expected TypedValue for Double type of result value" in {
+  "buildMeasurementDTO" should "build MeasurementDTO for BigDecimal type of result value when Measurement provided" in {
     val measure = SumOfValuesOfColumn("col")
-    val measurement = Measurement(measure, 3.14)
+    val measurement = MeasurementProvided(measure, BigDecimal(3.14))
     val measurementDTO = MeasurementBuilder.buildMeasurementDTO(measurement)
 
-    val expectedTypedValue = TypedValue("3.14", ResultValueType.Double)
+    val expectedTypedValue = TypedValue("3.14", ResultValueType.BigDecimal)
 
     assert(measurementDTO.result.mainValue == expectedTypedValue)
   }
 
-  "buildMeasurementDTO" should "build MeasurementDTO with expected TypedValue for BigDecimal type of result value" in {
+  "buildMeasurementDTO" should "not build MeasurementDTO for incompatible String type of result value when Measurement provided" in {
     val measure = SumOfValuesOfColumn("col")
-    val measurement = Measurement(measure, BigDecimal(3.14))
+    val measurement = MeasurementByAtum(measure, "stringValue", ResultValueType.String)
+
     val measurementDTO = MeasurementBuilder.buildMeasurementDTO(measurement)
 
-    val expectedTypedValue = TypedValue("3.14", ResultValueType.BigDecimal)
+    val expectedTypedValue = TypedValue("stringValue", ResultValueType.String)
 
     assert(measurementDTO.result.mainValue == expectedTypedValue)
   }
 
-  "buildMeasurementDTO" should "build MeasurementDTO with expected TypedValue for String type of result value" in {
+  "buildMeasurementDTO" should "build MeasurementDTO for BigDecimal type of result value when measured by Agent" in {
     val measure = SumOfValuesOfColumn("col")
-    val measurement = Measurement(measure, "stringValue")
+    val measurement = MeasurementByAtum(measure, "1", ResultValueType.BigDecimal)
     val measurementDTO = MeasurementBuilder.buildMeasurementDTO(measurement)
 
-    val expectedTypedValue = TypedValue("stringValue", ResultValueType.String)
-
-    assert(measurementDTO.result.mainValue == expectedTypedValue)
-  }
+    val expectedMeasureDTO = MeasureDTO("aggregatedTotal", Seq("col"))
 
-  "buildMeasurementDTO" should "throw exception for unsupported result value type" in {
-    val measure = SumOfValuesOfColumn("col")
-    val measurement = Measurement(measure, 1)
+    val expectedMeasureResultDTO = MeasureResultDTO(
+      TypedValue("1", ResultValueType.BigDecimal)
+    )
 
-    assertThrows[UnsupportedMeasureResultType](MeasurementBuilder.buildMeasurementDTO(measurement))
+    assert(measurementDTO.measure == expectedMeasureDTO)
+    assert(measurementDTO.result == expectedMeasureResultDTO)
   }
 
 }

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

@@ -0,0 +1,61 @@
+/*
+ * 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.agent.model
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import za.co.absa.atum.agent.exception.MeasurementProvidedException
+import za.co.absa.atum.agent.model.Measure._
+import za.co.absa.atum.agent.model.Measurement.MeasurementProvided
+import za.co.absa.atum.model.dto.MeasureResultDTO.ResultValueType
+import za.co.absa.spark.commons.test.SparkTestBase
+
+class MeasurementTest extends AnyFlatSpec with Matchers with SparkTestBase { self =>
+
+  "MeasurementProvided" should "be able to be converted to MeasurementProvided object when the result is Double" in {
+    val measure = AbsSumOfValuesOfColumn("col")
+    val actualMeasurement = MeasurementProvided(measure, 1.0)
+
+    assert(actualMeasurement.resultValue == 1.0)
+    assert(actualMeasurement.resultType == ResultValueType.Double)
+  }
+
+  "MeasurementProvided" should "throw exception for unsupported result value - BigDecimal instead of Double" in {
+    val measure = AbsSumOfValuesOfColumn("col")
+    assertThrows[MeasurementProvidedException](MeasurementProvided(measure, BigDecimal(1.0)))
+  }
+
+  "MeasurementProvided" should "throw exception for unsupported result value type in general (scalar)" in {
+    val measure = SumOfValuesOfColumn("col")
+    assertThrows[MeasurementProvidedException](MeasurementProvided(measure, 1))
+  }
+
+  "MeasurementProvided" should "throw exception for unsupported result value type in general (composite)" in {
+    val measure = SumOfHashesOfColumn("col")
+    assertThrows[MeasurementProvidedException](MeasurementProvided(measure, Map(1 -> "no-go")))
+  }
+
+  "MeasurementProvided" should "throw exception for unsupported result value type for a given Measure" in {
+    val measure = DistinctRecordCount("col")
+    assertThrows[MeasurementProvidedException](MeasurementProvided(measure, "1"))
+  }
+
+  "MeasurementProvided" should "throw exception for unsupported (slightly different FPN) result value type for a given Measure" in {
+    val measure = SumOfValuesOfColumn("col")
+    assertThrows[MeasurementProvidedException](MeasurementProvided(measure, 1.0))
+  }
+}

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

@@ -17,11 +17,11 @@
 package za.co.absa.atum.agent.model
 
 import org.scalatest.flatspec.AnyFlatSpecLike
-import za.co.absa.atum.agent.exception.UnsupportedMeasureException
-import za.co.absa.atum.agent.model.Measure.{AbsSumOfValuesOfColumn, DistinctRecordCount, RecordCount, SumOfHashesOfColumn, SumOfValuesOfColumn}
+import za.co.absa.atum.agent.exception.MeasureException
+import za.co.absa.atum.agent.model.Measure._
 import za.co.absa.atum.model.dto.MeasureDTO
 
-class MeasuresMapperTest extends AnyFlatSpecLike {
+class MeasuresBuilderTest extends AnyFlatSpecLike {
 
   "mapToMeasures" should "map MeasureDTO into Measure for supported measures" in {
     val supportedMeasures = Set(
@@ -40,7 +40,7 @@ class MeasuresMapperTest extends AnyFlatSpecLike {
       SumOfHashesOfColumn("hashCrc32Col")
     )
 
-    val actualMeasures = MeasuresMapper.mapToMeasures(supportedMeasures)
+    val actualMeasures = MeasuresBuilder.mapToMeasures(supportedMeasures)
 
     assert(expectedMeasures == actualMeasures)
   }
@@ -50,7 +50,7 @@ class MeasuresMapperTest extends AnyFlatSpecLike {
       MeasureDTO("unsupportedMeasure", Seq("col"))
     )
 
-    assertThrows[UnsupportedMeasureException](MeasuresMapper.mapToMeasures(unsupportedMeasure))
+    assertThrows[MeasureException](MeasuresBuilder.mapToMeasures(unsupportedMeasure))
   }
 
 }

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

@@ -17,6 +17,6 @@
 package za.co.absa.atum.model.dto
 
 case class MeasureDTO(
-  functionName: String,
+  measureName: String,
   controlColumns: Seq[String]
 )

agent/src/test/scala/za/co/absa/atum/agent/model/MeasuresMapperTest.scala agent/src/test/scala/za/co/absa/atum/agent/model/MeasuresBuilderTest.scala

@@ -82,14 +82,14 @@ class SerializationUtilsTest extends AnyFlatSpecLike {
       measures = seqMeasureDTO
     )
 
-    val expectedAdditionalDataJson = "{\"partitioning\":[{\"key\":\"key\",\"value\":\"val\"}],\"measures\":[{\"functionName\":\"count\",\"controlColumns\":[\"col\"]}],\"additionalData\":{\"additionalData\":{}}}"
+    val expectedAdditionalDataJson = "{\"partitioning\":[{\"key\":\"key\",\"value\":\"val\"}],\"measures\":[{\"measureName\":\"count\",\"controlColumns\":[\"col\"]}],\"additionalData\":{\"additionalData\":{}}}"
     val actualAdditionalDataJson = SerializationUtils.asJson(atumContextDTO)
 
     assert(expectedAdditionalDataJson == actualAdditionalDataJson)
   }
 
   "fromJson" should "deserialize AtumContextDTO from json string" in {
-    val atumContextDTOJson = "{\"partitioning\":[{\"key\":\"key\",\"value\":\"val\"}],\"measures\":[{\"functionName\":\"count\",\"controlColumns\":[\"col\"]}],\"additionalData\":{\"additionalData\":{}}}"
+    val atumContextDTOJson = "{\"partitioning\":[{\"key\":\"key\",\"value\":\"val\"}],\"measures\":[{\"measureName\":\"count\",\"controlColumns\":[\"col\"]}],\"additionalData\":{\"additionalData\":{}}}"
 
     val seqPartitionDTO = Seq(PartitionDTO("key", "val"))
     val seqMeasureDTO = Set(MeasureDTO("count", Seq("col")))
@@ -150,7 +150,7 @@ class SerializationUtilsTest extends AnyFlatSpecLike {
       measurements = seqMeasurementDTO
     )
 
-    val expectedCheckpointDTOJson = "{\"id\":\"" + uuid + "\",\"name\":\"checkpoint\",\"author\":\"author\",\"measuredByAtumAgent\":true,\"partitioning\":[{\"key\":\"key\",\"value\":\"val\"}],\"processStartTime\":\"2023-10-24 10:20:59.005000+02\",\"processEndTime\":\"2023-10-24 10:20:59.005000+02\",\"measurements\":[{\"measure\":{\"functionName\":\"count\",\"controlColumns\":[\"col\"]},\"result\":{\"mainValue\":{\"value\":\"1\",\"valueType\":\"Long\"},\"supportValues\":{}}}]}"
+    val expectedCheckpointDTOJson = "{\"id\":\"" + uuid + "\",\"name\":\"checkpoint\",\"author\":\"author\",\"measuredByAtumAgent\":true,\"partitioning\":[{\"key\":\"key\",\"value\":\"val\"}],\"processStartTime\":\"2023-10-24 10:20:59.005000+02\",\"processEndTime\":\"2023-10-24 10:20:59.005000+02\",\"measurements\":[{\"measure\":{\"measureName\":\"count\",\"controlColumns\":[\"col\"]},\"result\":{\"mainValue\":{\"value\":\"1\",\"valueType\":\"Long\"},\"supportValues\":{}}}]}"
     val actualCheckpointDTOJson = SerializationUtils.asJson(checkpointDTO)
 
     assert(expectedCheckpointDTOJson == actualCheckpointDTOJson)
@@ -161,7 +161,7 @@ class SerializationUtilsTest extends AnyFlatSpecLike {
     val seqPartitionDTO = Seq(PartitionDTO("key", "val"))
     val timeWithZone = OffsetDateTime.of(2023, 10, 24, 10, 20, 59, 5000000, ZoneOffset.ofHours(2))
 
-    val checkpointDTOJson = "{\"id\":\"" + uuid + "\",\"name\":\"checkpoint\",\"author\":\"author\",\"measuredByAtumAgent\":true,\"partitioning\":[{\"key\":\"key\",\"value\":\"val\"}],\"processStartTime\":\"2023-10-24 10:20:59.005000+02\",\"processEndTime\":\"2023-10-24 10:20:59.005000+02\",\"measurements\":[{\"measure\":{\"functionName\":\"count\",\"controlColumns\":[\"col\"]},\"result\":{\"mainValue\":{\"value\":\"1\",\"valueType\":\"Long\"},\"supportValues\":{}}}]}"
+    val checkpointDTOJson = "{\"id\":\"" + uuid + "\",\"name\":\"checkpoint\",\"author\":\"author\",\"measuredByAtumAgent\":true,\"partitioning\":[{\"key\":\"key\",\"value\":\"val\"}],\"processStartTime\":\"2023-10-24 10:20:59.005000+02\",\"processEndTime\":\"2023-10-24 10:20:59.005000+02\",\"measurements\":[{\"measure\":{\"measureName\":\"count\",\"controlColumns\":[\"col\"]},\"result\":{\"mainValue\":{\"value\":\"1\",\"valueType\":\"Long\"},\"supportValues\":{}}}]}"
 
     val seqMeasurementDTO = Seq(
       MeasurementDTO(
@@ -191,14 +191,14 @@ class SerializationUtilsTest extends AnyFlatSpecLike {
   "asJson" should "serialize MeasureDTO into json string" in {
     val measureDTO = MeasureDTO("count", Seq("col"))
 
-    val expectedMeasureDTOJson = "{\"functionName\":\"count\",\"controlColumns\":[\"col\"]}"
+    val expectedMeasureDTOJson = "{\"measureName\":\"count\",\"controlColumns\":[\"col\"]}"
     val actualMeasureDTOJson = SerializationUtils.asJson(measureDTO)
 
     assert(expectedMeasureDTOJson == actualMeasureDTOJson)
   }
 
   "fromJson" should "deserialize MeasureDTO from json string" in {
-    val measureDTOJson = "{\"functionName\":\"count\",\"controlColumns\":[\"col\"]}"
+    val measureDTOJson = "{\"measureName\":\"count\",\"controlColumns\":[\"col\"]}"
 
     val expectedMeasureDTO = MeasureDTO("count", Seq("col"))
     val actualMeasureDTO = SerializationUtils.fromJson[MeasureDTO](measureDTOJson)
@@ -213,14 +213,14 @@ class SerializationUtilsTest extends AnyFlatSpecLike {
 
     val measurementDTO = MeasurementDTO(measureDTO, measureResultDTO)
 
-    val expectedMeasurementDTOJson = "{\"measure\":{\"functionName\":\"count\",\"controlColumns\":[\"col\"]},\"result\":{\"mainValue\":{\"value\":\"1\",\"valueType\":\"Long\"},\"supportValues\":{}}}"
+    val expectedMeasurementDTOJson = "{\"measure\":{\"measureName\":\"count\",\"controlColumns\":[\"col\"]},\"result\":{\"mainValue\":{\"value\":\"1\",\"valueType\":\"Long\"},\"supportValues\":{}}}"
     val actualMeasurementDTOJson = SerializationUtils.asJson(measurementDTO)
 
     assert(expectedMeasurementDTOJson == actualMeasurementDTOJson)
   }
 
   "fromJson" should "deserialize MeasurementDTO from json string" in {
-    val measurementDTOJson = "{\"measure\":{\"functionName\":\"count\",\"controlColumns\":[\"col\"]},\"result\":{\"mainValue\":{\"value\":\"1\",\"valueType\":\"Long\"},\"supportValues\":{}}}"
+    val measurementDTOJson = "{\"measure\":{\"measureName\":\"count\",\"controlColumns\":[\"col\"]},\"result\":{\"mainValue\":{\"value\":\"1\",\"valueType\":\"Long\"},\"supportValues\":{}}}"
 
     val measureDTO = MeasureDTO("count", Seq("col"))
     val measureResultDTO = MeasureResultDTO(mainValue = TypedValue("1", ResultValueType.Long))