Merge pull request #117 from data-catering/feature/transformations

Update version to 0.17.2, introduce custom transformations for post-g…

Author: pflooky

.gitignore

@@ -56,6 +56,7 @@ example/build
 example/.gradle
 example/benchmark/build
 example/benchmark/jars
+.tmp_prev_class_name
 
 *.class
 *.log

Dockerfile

@@ -6,7 +6,7 @@ RUN addgroup -S app \
     && apk update --no-cache \
     && apk update --no-cache openssl \
     && apk add --no-cache libc6-compat bash \
-    && mkdir -p /opt/app /opt/DataCaterer/connection /opt/DataCaterer/plan /opt/DataCaterer/execution /opt/DataCaterer/report \
+    && mkdir -p /opt/app /opt/app/custom /opt/DataCaterer/connection /opt/DataCaterer/plan /opt/DataCaterer/execution /opt/DataCaterer/report \
     && chown -R app:app /opt/app /opt/DataCaterer/connection /opt/DataCaterer/plan /opt/DataCaterer/execution /opt/DataCaterer/report
 COPY --chown=app:app misc/docker-image app/src/main/resources app/build/libs /opt/app/
 

README.md

@@ -38,7 +38,7 @@ and deep dive into issues [from the generated report](https://data.catering/late
 
 1. Docker
    ```shell
-   docker run -d -i -p 9898:9898 -e DEPLOY_MODE=standalone --name datacaterer datacatering/data-caterer:0.17.1
+   docker run -d -i -p 9898:9898 -e DEPLOY_MODE=standalone --name datacaterer datacatering/data-caterer:0.17.2
    ```
    [Open localhost:9898](http://localhost:9898).
 1. [Run Scala/Java examples](#run-scalajava-examples)

api/src/main/scala/io/github/datacatering/datacaterer/api/TaskBuilder.scala

@@ -5,7 +5,7 @@ import io.github.datacatering.datacaterer.api.HttpMethodEnum.HttpMethodEnum
 import io.github.datacatering.datacaterer.api.HttpQueryParameterStyleEnum.HttpQueryParameterStyleEnum
 import io.github.datacatering.datacaterer.api.converter.Converters.{toScalaList, toScalaMap}
 import io.github.datacatering.datacaterer.api.model.Constants._
-import io.github.datacatering.datacaterer.api.model.{ArrayType, Count, DataType, DoubleType, Field, HeaderType, PerFieldCount, Step, StringType, Task, TaskSummary}
+import io.github.datacatering.datacaterer.api.model.{ArrayType, Count, DataType, DoubleType, Field, HeaderType, PerFieldCount, Step, StringType, Task, TaskSummary, TransformationConfig}
 
 import scala.annotation.varargs
 import scala.collection.JavaConverters._
@@ -168,6 +168,88 @@ case class TaskBuilder(task: Task = Task()) {
     )
     this.modify(_.task.steps)(_ ++ List(placeholderStep))
   }
+
+  /**
+   * Configure custom transformation for all steps in this task. Defaults to whole-file mode.
+   * Transformations execute after files are written ("last mile" transformation).
+   * Step-level transformations will override this task-level configuration.
+   *
+   * @param className Fully qualified class name of the transformer
+   * @return TaskBuilder
+   */
+  def transformation(className: String): TaskBuilder =
+    this.modify(_.task.transformation).setTo(Some(TransformationConfig(className = className)))
+
+  /**
+   * Configure per-record transformation for all steps. Transforms each record/line individually.
+   *
+   * @param className  Fully qualified class name of the transformer
+   * @param methodName Method name to invoke (default: "transformRecord")
+   * @return TaskBuilder
+   */
+  def transformationPerRecord(className: String, methodName: String = "transformRecord"): TaskBuilder =
+    this.modify(_.task.transformation).setTo(Some(TransformationConfig(
+      className = className,
+      methodName = methodName,
+      mode = "per-record"
+    )))
+
+  /**
+   * Configure whole-file transformation for all steps. Transforms entire files as units.
+   *
+   * @param className  Fully qualified class name of the transformer
+   * @param methodName Method name to invoke (default: "transformFile")
+   * @return TaskBuilder
+   */
+  def transformationWholeFile(className: String, methodName: String = "transformFile"): TaskBuilder =
+    this.modify(_.task.transformation).setTo(Some(TransformationConfig(
+      className = className,
+      methodName = methodName,
+      mode = "whole-file"
+    )))
+
+  /**
+   * Set output path for transformation. If specified, creates new files instead of replacing originals.
+   *
+   * @param outputPath     Path where transformed output should be written
+   * @param deleteOriginal Whether to delete original files after transformation (default: false)
+   * @return TaskBuilder
+   */
+  def transformationOutput(outputPath: String, deleteOriginal: Boolean = false): TaskBuilder =
+    this.task.transformation match {
+      case Some(config) =>
+        this.modify(_.task.transformation).setTo(Some(config.copy(outputPath = Some(outputPath), deleteOriginal = deleteOriginal)))
+      case None =>
+        this
+    }
+
+  /**
+   * Add options to transformation configuration.
+   *
+   * @param options Options to pass to the transformer
+   * @return TaskBuilder
+   */
+  def transformationOptions(options: Map[String, String]): TaskBuilder =
+    this.task.transformation match {
+      case Some(config) =>
+        this.modify(_.task.transformation).setTo(Some(config.copy(options = config.options ++ options)))
+      case None =>
+        this
+    }
+
+  /**
+   * Explicitly enable or disable transformation for all steps in this task.
+   *
+   * @param enabled Whether transformation is enabled
+   * @return TaskBuilder
+   */
+  def enableTransformation(enabled: Boolean): TaskBuilder =
+    this.task.transformation match {
+      case Some(config) =>
+        this.modify(_.task.transformation).setTo(Some(config.copy(enabled = enabled)))
+      case None =>
+        this
+    }
 }
 
 case class StepBuilder(step: Step = Step(), optValidation: Option[DataSourceValidationBuilder] = None) {
@@ -490,6 +572,87 @@ case class StepBuilder(step: Step = Step(), optValidation: Option[DataSourceVali
   @varargs def excludeFieldPatterns(patterns: String*): StepBuilder =
     this.modify(_.step.options)(_ ++ Map(EXCLUDE_FIELD_PATTERNS -> patterns.mkString(",")))
 
+  /**
+   * Configure custom transformation for this step. Defaults to whole-file mode.
+   * Transformations execute after the file is written ("last mile" transformation).
+   *
+   * @param className Fully qualified class name of the transformer
+   * @return StepBuilder
+   */
+  def transformation(className: String): StepBuilder =
+    this.modify(_.step.transformation).setTo(Some(TransformationConfig(className = className)))
+
+  /**
+   * Configure per-record transformation. Transforms each record/line in the file individually.
+   *
+   * @param className  Fully qualified class name of the transformer
+   * @param methodName Method name to invoke (default: "transformRecord")
+   * @return StepBuilder
+   */
+  def transformationPerRecord(className: String, methodName: String = "transformRecord"): StepBuilder =
+    this.modify(_.step.transformation).setTo(Some(TransformationConfig(
+      className = className,
+      methodName = methodName,
+      mode = "per-record"
+    )))
+
+  /**
+   * Configure whole-file transformation. Transforms the entire file as a unit.
+   *
+   * @param className  Fully qualified class name of the transformer
+   * @param methodName Method name to invoke (default: "transformFile")
+   * @return StepBuilder
+   */
+  def transformationWholeFile(className: String, methodName: String = "transformFile"): StepBuilder =
+    this.modify(_.step.transformation).setTo(Some(TransformationConfig(
+      className = className,
+      methodName = methodName,
+      mode = "whole-file"
+    )))
+
+  /**
+   * Set output path for transformation. If specified, creates a new file instead of replacing the original.
+   *
+   * @param outputPath     Path where transformed output should be written
+   * @param deleteOriginal Whether to delete the original file after transformation (default: false)
+   * @return StepBuilder
+   */
+  def transformationOutput(outputPath: String, deleteOriginal: Boolean = false): StepBuilder =
+    this.step.transformation match {
+      case Some(config) =>
+        this.modify(_.step.transformation).setTo(Some(config.copy(outputPath = Some(outputPath), deleteOriginal = deleteOriginal)))
+      case None =>
+        this
+    }
+
+  /**
+   * Add options to transformation configuration.
+   *
+   * @param options Options to pass to the transformer
+   * @return StepBuilder
+   */
+  def transformationOptions(options: Map[String, String]): StepBuilder =
+    this.step.transformation match {
+      case Some(config) =>
+        this.modify(_.step.transformation).setTo(Some(config.copy(options = config.options ++ options)))
+      case None =>
+        this
+    }
+
+  /**
+   * Explicitly enable or disable transformation for this step.
+   *
+   * @param enabled Whether transformation is enabled
+   * @return StepBuilder
+   */
+  def enableTransformation(enabled: Boolean): StepBuilder =
+    this.step.transformation match {
+      case Some(config) =>
+        this.modify(_.step.transformation).setTo(Some(config.copy(enabled = enabled)))
+      case None =>
+        this
+    }
+
   private def getValidation: DataSourceValidationBuilder = optValidation.getOrElse(DataSourceValidationBuilder())
 }
 

api/src/main/scala/io/github/datacatering/datacaterer/api/connection/ConnectionBuilder.scala

@@ -294,6 +294,76 @@ trait ConnectionTaskBuilder[T] {
     this
   }
 
+  /**
+   * Configure a transformation to apply to the generated data.
+   * By default, uses whole-file mode with method name "transformFile".
+   *
+   * @param className The fully qualified class name of the transformer
+   * @return The connection task builder
+   */
+  def transformation(className: String): ConnectionTaskBuilder[T] = {
+    this.step = Some(getStep.transformation(className))
+    this
+  }
+
+  /**
+   * Configure a per-record transformation to apply to the generated data.
+   *
+   * @param className  The fully qualified class name of the transformer
+   * @param methodName The method name to invoke (default: "transform")
+   * @return The connection task builder
+   */
+  def transformationPerRecord(className: String, methodName: String = "transform"): ConnectionTaskBuilder[T] = {
+    this.step = Some(getStep.transformationPerRecord(className, methodName))
+    this
+  }
+
+  /**
+   * Configure a whole-file transformation to apply to the generated data.
+   *
+   * @param className  The fully qualified class name of the transformer
+   * @param methodName The method name to invoke (default: "transformFile")
+   * @return The connection task builder
+   */
+  def transformationWholeFile(className: String, methodName: String = "transformFile"): ConnectionTaskBuilder[T] = {
+    this.step = Some(getStep.transformationWholeFile(className, methodName))
+    this
+  }
+
+  /**
+   * Configure the output path and deletion behavior for transformation.
+   *
+   * @param outputPath     The path where the transformed file should be written
+   * @param deleteOriginal Whether to delete the original file after transformation (default: false)
+   * @return The connection task builder
+   */
+  def transformationOutput(outputPath: String, deleteOriginal: Boolean = false): ConnectionTaskBuilder[T] = {
+    this.step = Some(getStep.transformationOutput(outputPath, deleteOriginal))
+    this
+  }
+
+  /**
+   * Add options for the transformation.
+   *
+   * @param options Map of option key-value pairs
+   * @return The connection task builder
+   */
+  def transformationOptions(options: Map[String, String]): ConnectionTaskBuilder[T] = {
+    this.step = Some(getStep.transformationOptions(options))
+    this
+  }
+
+  /**
+   * Enable or disable the transformation.
+   *
+   * @param enabled Whether the transformation should be enabled
+   * @return The connection task builder
+   */
+  def enableTransformation(enabled: Boolean): ConnectionTaskBuilder[T] = {
+    this.step = Some(getStep.enableTransformation(enabled))
+    this
+  }
+
   def toTasksBuilder: Option[TasksBuilder] = {
     val dataSourceName = connectionConfigWithTaskBuilder.dataSourceName
     val format = connectionConfigWithTaskBuilder.options(FORMAT)

api/src/main/scala/io/github/datacatering/datacaterer/api/model/PlanModels.scala

@@ -47,7 +47,8 @@ case class TaskSummary(
 
 case class Task(
                  name: String = DEFAULT_TASK_NAME,
-                 steps: List[Step] = List()
+                 steps: List[Step] = List(),
+                 transformation: Option[TransformationConfig] = None
                )
 
 case class Step(
@@ -56,7 +57,8 @@ case class Step(
                  count: Count = Count(),
                  options: Map[String, String] = Map(),
                  fields: List[Field] = List(),
-                 enabled: Boolean = DEFAULT_STEP_ENABLED
+                 enabled: Boolean = DEFAULT_STEP_ENABLED,
+                 transformation: Option[TransformationConfig] = None
                )
 
 case class Count(

api/src/main/scala/io/github/datacatering/datacaterer/api/model/TransformationModels.scala

@@ -0,0 +1,37 @@
+package io.github.datacatering.datacaterer.api.model
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties
+
+/**
+ * Configuration for custom transformation of generated data.
+ * Transformations can operate in two modes:
+ * - "per-record": Transform each record/line in the file individually
+ * - "whole-file": Transform the entire file as a unit
+ *
+ * Both modes execute after the file has been written ("last mile" transformation).
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+case class TransformationConfig(
+                                 className: String = "",
+                                 methodName: String = "transform",
+                                 mode: String = "whole-file",  // "per-record" or "whole-file"
+                                 outputPath: Option[String] = None,  // Output path (if different from input)
+                                 deleteOriginal: Boolean = false,  // Delete original file after transformation
+                                 options: Map[String, String] = Map(),
+                                 enabled: Boolean = true
+                               ) {
+  /**
+   * Check if transformation is configured and enabled
+   */
+  def isEnabled: Boolean = className.nonEmpty && enabled
+
+  /**
+   * Check if mode is per-record
+   */
+  def isPerRecord: Boolean = mode.equalsIgnoreCase("per-record")
+
+  /**
+   * Check if mode is whole-file
+   */
+  def isWholeFile: Boolean = mode.equalsIgnoreCase("whole-file")
+}