Merge pull request #110 from data-catering/feature/sample-yaml-metadata

Feature/sample yaml metadata

Author: pflooky

.gitignore

@@ -32,6 +32,9 @@ app/src/test/resources/sample/documentation
 app/src/test/resources/sample/java
 app/src/test/resources/sample/report
 app/src/test/resources/sample/plan-gen
+app/src/test/resources/sample/e2e/generated
+app/src/test/resources/sample/e2e/report
+app/src/test/resources/sample/e2e/data
 
 api/out
 api/src/test/resources/sample/documentation

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.16.11
+   docker run -d -i -p 9898:9898 -e DEPLOY_MODE=standalone --name datacaterer datacatering/data-caterer:0.17.0
    ```
    [Open localhost:9898](http://localhost:9898).
 1. [Run Scala/Java examples](#run-scalajava-examples)

api/src/main/java/io/github/datacatering/datacaterer/javaapi/api/PlanRun.java

@@ -124,6 +124,18 @@ public GeneratorBuilder generator() {
         return new GeneratorBuilder();
     }
 
+    /**
+     * Creates a WeightedValue for weighted field values. Useful for oneOfWeightedJava field method.
+     *
+     * @param value  The value
+     * @param weight The weight associated with the value
+     * @return A WeightedValue containing the value and weight
+     */
+    public static WeightedValue weightedValue(Object value, double weight) {
+        return new WeightedValue(value, weight);
+    }
+
+
     /**
      * Creates a CountBuilder instance.
      *
@@ -819,6 +831,21 @@ public void execute(
         execute(plan(), configuration(), Collections.emptyList(), connectionTaskBuilder, connectionTaskBuilders);
     }
 
+    /**
+     * Executes the data catering plan with the provided list of tasks.
+     *
+     * @param connectionTaskBuilders A list of ConnectionTaskBuilder objects representing tasks to be executed.
+     */
+    public void execute(List<ConnectionTaskBuilder<?>> connectionTaskBuilders) {
+        if (connectionTaskBuilders.isEmpty()) {
+            throw new IllegalArgumentException("At least one ConnectionTaskBuilder must be provided");
+        }
+        ConnectionTaskBuilder<?> first = connectionTaskBuilders.get(0);
+        ConnectionTaskBuilder<?>[] rest = connectionTaskBuilders.subList(1, connectionTaskBuilders.size())
+                .toArray(new ConnectionTaskBuilder<?>[0]);
+        execute(first, rest);
+    }
+
 
     /**
      * Executes the data catering plan with the provided configurations.
@@ -835,6 +862,25 @@ public void execute(
         execute(plan(), configurationBuilder, Collections.emptyList(), connectionTaskBuilder, connectionTaskBuilders);
     }
 
+    /**
+     * Executes the data catering plan with the provided configuration and list of tasks.
+     *
+     * @param configurationBuilder   The DataCatererConfigurationBuilder object representing the configuration for the plan execution.
+     * @param connectionTaskBuilders A list of ConnectionTaskBuilder objects representing tasks to be executed.
+     */
+    public void execute(
+            DataCatererConfigurationBuilder configurationBuilder,
+            List<ConnectionTaskBuilder<?>> connectionTaskBuilders
+    ) {
+        if (connectionTaskBuilders.isEmpty()) {
+            throw new IllegalArgumentException("At least one ConnectionTaskBuilder must be provided");
+        }
+        ConnectionTaskBuilder<?> first = connectionTaskBuilders.get(0);
+        ConnectionTaskBuilder<?>[] rest = connectionTaskBuilders.subList(1, connectionTaskBuilders.size())
+                .toArray(new ConnectionTaskBuilder<?>[0]);
+        execute(configurationBuilder, first, rest);
+    }
+
     /**
      * Executes the data catering plan with the provided configurations.
      *
@@ -852,6 +898,27 @@ public void execute(
         execute(planBuilder, configurationBuilder, Collections.emptyList(), connectionTaskBuilder, connectionTaskBuilders);
     }
 
+    /**
+     * Executes the data catering plan with the provided plan, configuration and list of tasks.
+     *
+     * @param planBuilder            The PlanBuilder object representing the plan to be executed.
+     * @param configurationBuilder   The DataCatererConfigurationBuilder object representing the configuration for the plan execution.
+     * @param connectionTaskBuilders A list of ConnectionTaskBuilder objects representing tasks to be executed.
+     */
+    public void execute(
+            PlanBuilder planBuilder,
+            DataCatererConfigurationBuilder configurationBuilder,
+            List<ConnectionTaskBuilder<?>> connectionTaskBuilders
+    ) {
+        if (connectionTaskBuilders.isEmpty()) {
+            throw new IllegalArgumentException("At least one ConnectionTaskBuilder must be provided");
+        }
+        ConnectionTaskBuilder<?> first = connectionTaskBuilders.get(0);
+        ConnectionTaskBuilder<?>[] rest = connectionTaskBuilders.subList(1, connectionTaskBuilders.size())
+                .toArray(new ConnectionTaskBuilder<?>[0]);
+        execute(planBuilder, configurationBuilder, first, rest);
+    }
+
     /**
      * Executes the data catering plan with the provided configurations.
      *

api/src/main/java/io/github/datacatering/datacaterer/javaapi/api/WeightedValue.java

@@ -0,0 +1,49 @@
+package io.github.datacatering.datacaterer.javaapi.api;
+
+/**
+ * A simple container for value-weight pairs used in Java API for oneOfWeighted field method.
+ */
+public class WeightedValue {
+    private final Object value;
+    private final double weight;
+
+    /**
+     * Constructs a new WeightedValue.
+     * 
+     * @param value  The value
+     * @param weight The weight associated with the value
+     */
+    public WeightedValue(Object value, double weight) {
+        this.value = value;
+        this.weight = weight;
+    }
+
+    /**
+     * Gets the value.
+     * 
+     * @return The value
+     */
+    public Object value() {
+        return value;
+    }
+
+    /**
+     * Gets the weight.
+     * 
+     * @return The weight
+     */
+    public double weight() {
+        return weight;
+    }
+
+    /**
+     * Static factory method for easier creation.
+     * 
+     * @param value  The value
+     * @param weight The weight associated with the value
+     * @return A new WeightedValue instance
+     */
+    public static WeightedValue of(Object value, double weight) {
+        return new WeightedValue(value, weight);
+    }
+}

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

@@ -2,8 +2,8 @@ package io.github.datacatering.datacaterer.api
 
 import com.softwaremill.quicklens.ModifyPimp
 import io.github.datacatering.datacaterer.api.converter.Converters.toScalaMap
-import io.github.datacatering.datacaterer.api.model.Constants.{CONFLUENT_SCHEMA_REGISTRY_ID, CONFLUENT_SCHEMA_REGISTRY_SUBJECT, CONFLUENT_SCHEMA_REGISTRY_VERSION, DATA_CONTRACT_FILE, DATA_CONTRACT_SCHEMA, GREAT_EXPECTATIONS_FILE, JSON_SCHEMA_FILE, METADATA_SOURCE_URL, OPEN_LINEAGE_DATASET, OPEN_LINEAGE_NAMESPACE, OPEN_METADATA_API_VERSION, OPEN_METADATA_AUTH_TYPE, OPEN_METADATA_AUTH_TYPE_OPEN_METADATA, OPEN_METADATA_DEFAULT_API_VERSION, OPEN_METADATA_HOST, OPEN_METADATA_JWT_TOKEN, SCHEMA_LOCATION}
-import io.github.datacatering.datacaterer.api.model.{ConfluentSchemaRegistrySource, DataContractCliSource, GreatExpectationsSource, JsonSchemaSource, MarquezMetadataSource, MetadataSource, OpenAPISource, OpenDataContractStandardSource, OpenMetadataSource}
+import io.github.datacatering.datacaterer.api.model.Constants.{CONFLUENT_SCHEMA_REGISTRY_ID, CONFLUENT_SCHEMA_REGISTRY_SUBJECT, CONFLUENT_SCHEMA_REGISTRY_VERSION, DATA_CONTRACT_FILE, DATA_CONTRACT_SCHEMA, GREAT_EXPECTATIONS_FILE, JSON_SCHEMA_FILE, METADATA_SOURCE_URL, OPEN_LINEAGE_DATASET, OPEN_LINEAGE_NAMESPACE, OPEN_METADATA_API_VERSION, OPEN_METADATA_AUTH_TYPE, OPEN_METADATA_AUTH_TYPE_OPEN_METADATA, OPEN_METADATA_DEFAULT_API_VERSION, OPEN_METADATA_HOST, OPEN_METADATA_JWT_TOKEN, SCHEMA_LOCATION, YAML_PLAN_FILE, YAML_STEP_NAME, YAML_TASK_FILE, YAML_TASK_NAME}
+import io.github.datacatering.datacaterer.api.model.{ConfluentSchemaRegistrySource, DataContractCliSource, GreatExpectationsSource, JsonSchemaSource, MarquezMetadataSource, MetadataSource, OpenAPISource, OpenDataContractStandardSource, OpenMetadataSource, YamlPlanSource, YamlTaskSource}
 
 case class MetadataSourceBuilder(metadataSource: MetadataSource = MarquezMetadataSource()) {
   def this() = this(MarquezMetadataSource())
@@ -138,4 +138,56 @@ case class MetadataSourceBuilder(metadataSource: MetadataSource = MarquezMetadat
 
   def jsonSchemaJava(schemaFile: String, options: java.util.Map[String, String]): MetadataSourceBuilder =
     jsonSchema(schemaFile, toScalaMap(options))
+
+  /**
+   * Create metadata source from YAML plan file. This allows referencing existing YAML plan files
+   * and optionally override configurations using the builder pattern.
+   *
+   * @param planFile Path to the YAML plan file
+   * @return MetadataSourceBuilder with YamlPlanSource
+   */
+  def yamlPlan(planFile: String): MetadataSourceBuilder = {
+    this.modify(_.metadataSource).setTo(YamlPlanSource(Map(YAML_PLAN_FILE -> planFile)))
+  }
+
+  /**
+   * Create metadata source from YAML task file. This allows referencing existing YAML task files
+   * and optionally override configurations using the builder pattern.
+   *
+   * @param taskFile Path to the YAML task file
+   * @return MetadataSourceBuilder with YamlTaskSource
+   */
+  def yamlTask(taskFile: String): MetadataSourceBuilder = {
+    this.modify(_.metadataSource).setTo(YamlTaskSource(Map(YAML_TASK_FILE -> taskFile)))
+  }
+
+  /**
+   * Create metadata source from YAML task file with specific task name filter.
+   *
+   * @param taskFile Path to the YAML task file
+   * @param taskName Name of the specific task to use from the YAML file
+   * @return MetadataSourceBuilder with YamlTaskSource
+   */
+  def yamlTask(taskFile: String, taskName: String): MetadataSourceBuilder = {
+    this.modify(_.metadataSource).setTo(YamlTaskSource(Map(
+      YAML_TASK_FILE -> taskFile,
+      YAML_TASK_NAME -> taskName
+    )))
+  }
+
+  /**
+   * Create metadata source from YAML task file with specific task and step name filters.
+   *
+   * @param taskFile Path to the YAML task file
+   * @param taskName Name of the specific task to use from the YAML file
+   * @param stepName Name of the specific step to use from the task
+   * @return MetadataSourceBuilder with YamlTaskSource
+   */
+  def yamlTask(taskFile: String, taskName: String, stepName: String): MetadataSourceBuilder = {
+    this.modify(_.metadataSource).setTo(YamlTaskSource(Map(
+      YAML_TASK_FILE -> taskFile,
+      YAML_TASK_NAME -> taskName,
+      YAML_STEP_NAME -> stepName
+    )))
+  }
 }

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

@@ -4,8 +4,8 @@ import com.softwaremill.quicklens.ModifyPimp
 import io.github.datacatering.datacaterer.api
 import io.github.datacatering.datacaterer.api.connection.ConnectionTaskBuilder
 import io.github.datacatering.datacaterer.api.converter.Converters.toScalaList
-import io.github.datacatering.datacaterer.api.model.Constants.{ENABLE_REFERENCE_MODE, METADATA_SOURCE_TYPE}
-import io.github.datacatering.datacaterer.api.model.{ForeignKeyRelation, Plan}
+import io.github.datacatering.datacaterer.api.model.Constants.{ENABLE_REFERENCE_MODE, METADATA_SOURCE_TYPE, YAML_PLAN}
+import io.github.datacatering.datacaterer.api.model.{ForeignKeyRelation, Plan, YamlPlanSource}
 
 import scala.annotation.varargs
 
@@ -146,4 +146,20 @@ case class PlanBuilder(plan: Plan = Plan(), tasks: List[TasksBuilder] = List())
       }
     })
   }
+
+  /**
+   * Create a PlanBuilder that loads base configuration from a YAML file.
+   * This allows referencing existing YAML plan definitions while still being able to override
+   * specific configurations using the builder pattern.
+   *
+   * @param yamlConfig Configuration specifying which YAML file to load
+   * @return PlanBuilder with YAML plan as base configuration
+   */
+  def fromYaml(yamlConfig: YamlConfig): PlanBuilder = {
+    // Add YAML source as metadata source type for later processing
+    val yamlSource = YamlPlanSource(yamlConfig.toOptionsMap + (METADATA_SOURCE_TYPE -> YAML_PLAN))
+    // Return current builder with special marker to indicate YAML loading
+    // The actual YAML loading will happen during execution when the plan is processed
+    this.modify(_.plan.description).setTo(s"${plan.description} [YAML: ${yamlConfig.planFile.getOrElse("unknown")}]")
+  }
 }

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

@@ -182,6 +182,14 @@ trait PlanRun {
    */
   def metadataSource: MetadataSourceBuilder = MetadataSourceBuilder()
 
+  /**
+   * Create new YAML builder for referencing existing YAML plan and task files.
+   * This allows loading existing YAML configurations and overriding specific parts using the programmatic API.
+   *
+   * @return YamlBuilder
+   */
+  def yaml: YamlBuilder = YamlBuilder()
+
   /**
    * Create new CSV generation step with configurations
    *
@@ -633,6 +641,26 @@ trait PlanRun {
     execute(configuration, connectionTaskBuilder, connectionTaskBuilders: _*)
   }
 
+  /**
+   * Execute with non-default configurations for a set of tasks
+   *
+   * @param baseConfiguration      Runtime configurations
+   * @param connectionTaskBuilders Connection and task builders
+   */
+  def execute(
+               baseConfiguration: DataCatererConfigurationBuilder,
+               connectionTaskBuilders: List[ConnectionTaskBuilder[_]]
+             ): Unit = {
+    if (connectionTaskBuilders.isEmpty) {
+      throw new IllegalArgumentException("At least one ConnectionTaskBuilder must be provided")
+    }
+    if (connectionTaskBuilders.tail.isEmpty) {
+      execute(plan, baseConfiguration, List(), connectionTaskBuilders.head)
+    } else {
+      execute(plan, baseConfiguration, List(), connectionTaskBuilders.head, connectionTaskBuilders.tail: _*)
+    }
+  }
+
   /**
    * Execute with non-default configurations for a set of tasks
    *
@@ -665,6 +693,28 @@ trait PlanRun {
     execute(planBuilder, baseConfiguration, List(), connectionTaskBuilder, connectionTaskBuilders: _*)
   }
 
+  /**
+   * Execute with non-default configurations with validations and tasks
+   *
+   * @param planBuilder            Plan to set high level task configurations
+   * @param baseConfiguration      Runtime configurations
+   * @param connectionTaskBuilders Connection and task builders
+   */
+  def execute(
+               planBuilder: PlanBuilder,
+               baseConfiguration: DataCatererConfigurationBuilder,
+               connectionTaskBuilders: List[ConnectionTaskBuilder[_]]
+             ): Unit = {
+    if (connectionTaskBuilders.isEmpty) {
+      throw new IllegalArgumentException("At least one ConnectionTaskBuilder must be provided")
+    }
+    if (connectionTaskBuilders.tail.isEmpty) {
+      execute(planBuilder, baseConfiguration, List(), connectionTaskBuilders.head)
+    } else {
+      execute(planBuilder, baseConfiguration, List(), connectionTaskBuilders.head, connectionTaskBuilders.tail: _*)
+    }
+  }
+
   /**
    * Execute with non-default configurations with validations and tasks. Validations have to be enabled before running
    * (see [[DataCatererConfigurationBuilder.enableValidation()]].