Merge pull request #108 from data-catering/feature/api-parse-yaml

Update version to 0.16.10, enhance YAML plan execution via UI/API, an…

Author: pflooky

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

api/build.gradle.kts

@@ -38,12 +38,10 @@ repositories {
 }
 
 val basicImpl: Configuration by configurations.creating
-val advancedImpl: Configuration by configurations.creating
 
 configurations {
     implementation {
         extendsFrom(basicImpl)
-        extendsFrom(advancedImpl)
     }
 }
 
@@ -77,7 +75,7 @@ testing {
 tasks.test {
     jvmArgs("-Djava.security.manager=allow", "-Djdk.module.illegalAccess=deny", "--add-opens=java.base/java.lang=ALL-UNNAMED", "--add-opens=java.base/java.lang.invoke=ALL-UNNAMED", "--add-opens=java.base/java.lang.reflect=ALL-UNNAMED", "--add-opens=java.base/java.io=ALL-UNNAMED", "--add-opens=java.base/java.net=ALL-UNNAMED", "--add-opens=java.base/java.nio=ALL-UNNAMED", "--add-opens=java.base/java.util=ALL-UNNAMED", "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED", "--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED", "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED", "--add-opens=java.base/sun.nio.cs=ALL-UNNAMED", "--add-opens=java.base/sun.security.action=ALL-UNNAMED", "--add-opens=java.base/sun.util.calendar=ALL-UNNAMED", "--add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED")
 
-    // Only finalize with Scoverage report when configuration cache is disabled
+    // Only finalize with Scoverage report when configuration cache is disable
     if (!isConfigCacheEnabled) {
         finalizedBy(tasks.named("reportScoverage"))
     }

app/build.gradle.kts

@@ -262,6 +262,15 @@ tasks.register<JavaExec>("runUI") {
     classpath = sourceSets.main.get().runtimeClasspath
     mainClass.set("io.github.datacatering.datacaterer.core.ui.DataCatererUI")
 
+    // Set environment variables for YAML plan and task paths (absolute paths)
+    val projectDir = project.projectDir.absolutePath
+    val rootProjectDir = project.rootProject.projectDir.absolutePath
+    environment("PLAN_FILE_PATH", "$rootProjectDir/app/src/test/resources/sample/plan/customer-create-plan.yaml")
+    environment("TASK_FOLDER_PATH", "$rootProjectDir/app/src/test/resources/sample/task")
+    
+    // Set working directory to project root to help with relative path resolution
+    workingDir(rootProjectDir)
+    
     // Add JVM arguments similar to docker script
     jvmArgs(
         "-Djava.security.manager=allow",

app/src/main/scala/io/github/datacatering/datacaterer/core/parser/PlanParser.scala

@@ -17,10 +17,10 @@ object PlanParser {
   private val LOGGER = Logger.getLogger(getClass.getName)
   private val OBJECT_MAPPER = ObjectMapperUtil.yamlObjectMapper
 
-  def getPlanTasksFromYaml(dataCatererConfiguration: DataCatererConfiguration)
+  def getPlanTasksFromYaml(dataCatererConfiguration: DataCatererConfiguration, enabledOnly: Boolean = true)
                           (implicit sparkSession: SparkSession): (Plan, List[Task], Option[List[ValidationConfiguration]]) = {
     val parsedPlan = PlanParser.parsePlan(dataCatererConfiguration.foldersConfig.planFilePath)
-    val enabledPlannedTasks = parsedPlan.tasks.filter(_.enabled)
+    val enabledPlannedTasks = if (enabledOnly) parsedPlan.tasks.filter(_.enabled) else parsedPlan.tasks
     val enabledTaskMap = enabledPlannedTasks.map(t => (t.name, t)).toMap
     val planWithEnabledTasks = parsedPlan.copy(tasks = enabledPlannedTasks)
 
@@ -40,7 +40,7 @@ object PlanParser {
       val planFile = FileUtil.getFile(planFilePath)
       OBJECT_MAPPER.readValue(planFile, classOf[Plan])
     }
-    LOGGER.info(s"Found plan file and parsed successfully, plan-file-path=$planFilePath, plan-name=${parsedPlan.name}, plan-description=${parsedPlan.description}")
+    LOGGER.debug(s"Found plan file and parsed successfully, plan-file-path=$planFilePath, plan-name=${parsedPlan.name}, plan-description=${parsedPlan.description}")
     parsedPlan
   }
 

app/src/main/scala/io/github/datacatering/datacaterer/core/plan/PlanProcessor.scala

@@ -12,9 +12,10 @@ import io.github.datacatering.datacaterer.core.model.Constants.METADATA_CONNECTI
 import io.github.datacatering.datacaterer.core.model.PlanRunResults
 import io.github.datacatering.datacaterer.core.parser.PlanParser
 import io.github.datacatering.datacaterer.core.util.SparkProvider
+import org.apache.log4j.Logger
 import org.apache.spark.sql.SparkSession
 
-import scala.util.{Success, Try}
+import scala.util.{Failure, Success, Try}
 
 object PlanProcessor {
 
@@ -103,7 +104,36 @@ object PlanProcessor {
   private def parsePlan(dataCatererConfiguration: DataCatererConfiguration, optPlan: Option[PlanRun], interface: String)(implicit sparkSession: SparkSession): (PlanRun, String) = {
     try {
       if (optPlan.isDefined) {
-        (optPlan.get, interface)
+        // Check if we need to load from YAML by looking at plan task summaries vs actual tasks
+        val existingTaskNames = optPlan.get._tasks.map(_.name).toSet
+        val planTaskNames = optPlan.get._plan.tasks.map(_.name).toSet
+        val missingTaskNames = planTaskNames.diff(existingTaskNames)
+        
+        if (missingTaskNames.nonEmpty) {
+          // Tasks are missing - this means the plan is from a YAML file, not the UI
+          // Load the entire plan and all tasks from the YAML file
+          val yamlConfig = ConfigParser.toDataCatererConfiguration
+          
+          // Find the correct YAML plan file by name in the configured plan directory
+          val requestedPlanName = optPlan.get._plan.name
+          val yamlPlanFilePath = findYamlPlanFile(yamlConfig.foldersConfig.planFilePath, requestedPlanName)
+          
+          // Load the plan from the specific YAML file if found, otherwise use default
+          val planConfigForParsing = yamlPlanFilePath.map { planPath =>
+            yamlConfig.copy(
+              foldersConfig = yamlConfig.foldersConfig.copy(planFilePath = planPath)
+            )
+          }.getOrElse(yamlConfig)
+          
+          // Load everything from YAML - use only YAML configuration, no UI data
+          val (parsedPlan, enabledTasks, validations) = PlanParser.getPlanTasksFromYaml(planConfigForParsing, false)
+          
+          val yamlPlanRun = new YamlPlanRun(parsedPlan, enabledTasks, validations, planConfigForParsing)
+          (yamlPlanRun, DATA_CATERER_INTERFACE_YAML)
+        } else {
+          // All tasks are provided by UI, this is a pure UI plan
+          (optPlan.get, interface)
+        }
       } else {
         val (parsedPlan, enabledTasks, validations) = PlanParser.getPlanTasksFromYaml(dataCatererConfiguration)
         (new YamlPlanRun(parsedPlan, enabledTasks, validations, dataCatererConfiguration), DATA_CATERER_INTERFACE_YAML)
@@ -114,6 +144,38 @@ object PlanProcessor {
         throw parsePlanException
     }
   }
+  
+  private def findYamlPlanFile(configuredPlanPath: String, planName: String)(implicit sparkSession: SparkSession): Option[String] = {
+    import java.io.File
+    
+    val logger = Logger.getLogger(getClass.getName)
+    
+    // Get the parent directory from the configured plan file path
+    val planFile = new File(configuredPlanPath)
+    val planDir = if (planFile.isDirectory) planFile else new File(planFile.getParent)
+    
+    if (planDir.exists() && planDir.isDirectory) {
+      // Look for a YAML file matching the plan name
+      val matchingFiles = planDir.listFiles()
+        .filter(f => f.isFile && f.getName.endsWith(".yaml"))
+        .filter(f => {
+          // Try to parse the plan and match by name
+          Try {
+            val parsed = PlanParser.parsePlan(f.getAbsolutePath)
+            parsed.name.equalsIgnoreCase(planName)
+          } match {
+            case Success(matches) => matches
+            case Failure(ex) =>
+              logger.warn(s"Failed to parse YAML plan file: ${f.getAbsolutePath}", ex)
+              false
+          }
+        })
+      
+      matchingFiles.headOption.map(_.getAbsolutePath)
+    } else {
+      None
+    }
+  }
 
   private def handleException(
                                 exception: Exception,
@@ -158,8 +220,8 @@ class YamlPlanRun(
                    dataCatererConfiguration: DataCatererConfiguration
                  ) extends PlanRun {
   _plan = yamlPlan
-  _tasks = yamlTasks
   _validations = validations.getOrElse(List())
+
   //get any metadata configuration from tasks for data sources and add to configuration
   private val tasksWithMetadataOptions = yamlTasks.filter(t => t.steps.nonEmpty)
     .map(t => {
@@ -168,5 +230,20 @@ class YamlPlanRun(
     }).toMap
   private val updatedConnectionConfig = dataCatererConfiguration.connectionConfigByName
     .map(c => c._1 -> (tasksWithMetadataOptions.getOrElse(c._1, Map()) ++ c._2))
+
+  // Merge connection configuration into task step options
+  // This ensures that step.options contains all necessary connection details like 'format'
+  _tasks = yamlTasks.map(task => {
+    val dataSourceName = yamlPlan.tasks.find(ts => ts.name.equalsIgnoreCase(task.name)).get.dataSourceName
+    val connectionConfig = updatedConnectionConfig.getOrElse(dataSourceName, Map())
+
+    // Merge connection config into each step's options (connection config as base, step options override)
+    val stepsWithConnectionConfig = task.steps.map(step => {
+      step.copy(options = connectionConfig ++ step.options)
+    })
+
+    task.copy(steps = stepsWithConnectionConfig)
+  })
+
   _configuration = dataCatererConfiguration.copy(connectionConfigByName = updatedConnectionConfig)
 }

app/src/main/scala/io/github/datacatering/datacaterer/core/ui/plan/ConnectionRepository.scala

@@ -1,5 +1,7 @@
 package io.github.datacatering.datacaterer.core.ui.plan
 
+import io.github.datacatering.datacaterer.api.model.Constants.FORMAT
+import io.github.datacatering.datacaterer.core.config.ConfigParser
 import io.github.datacatering.datacaterer.core.ui.config.UiConfiguration.INSTALL_DIRECTORY
 import io.github.datacatering.datacaterer.core.ui.model.{Connection, GetConnectionsResponse, SaveConnectionsRequest}
 import org.apache.log4j.Logger
@@ -60,19 +62,58 @@ object ConnectionRepository {
 
   def getConnection(name: String, masking: Boolean = true): Connection = {
     LOGGER.debug(s"Getting connection details, connection-name=$name")
+    
+    // First try to get from saved connections file
     val connectionFile = Path.of(s"$connectionSaveFolder/$name.csv")
-    val tryConnection = Try(Connection.fromString(Files.readString(connectionFile), name, masking))
-    tryConnection match {
+    val tryConnectionFromFile = if (connectionFile.toFile.exists()) {
+      Try(Connection.fromString(Files.readString(connectionFile), name, masking))
+    } else {
+      Failure(new IllegalArgumentException(s"Connection file not found: $name"))
+    }
+    
+    tryConnectionFromFile match {
       case Success(connection) => connection.copy(options = connection.options)
-      case Failure(exception) => throw exception
+      case Failure(fileException) =>
+        // If not found in file, try to get from application.conf
+        LOGGER.debug(s"Connection not found in file, checking application.conf, connection-name=$name")
+        Try(getConnectionFromConfig(name, masking)) match {
+          case Success(conn) => conn
+          case Failure(configException) =>
+            // If not found in either location, throw the original exception
+            throw new IllegalArgumentException(s"Connection not found: $name", fileException)
+        }
+    }
+  }
+  
+  /**
+   * Get connection details from application.conf
+   */
+  private def getConnectionFromConfig(name: String, masking: Boolean = true): Connection = {
+    val connectionConfigs = ConfigParser.connectionConfigsByName
+    connectionConfigs.get(name) match {
+      case Some(config) =>
+        val format = config.getOrElse(FORMAT, "unknown")
+        val options = if (masking) {
+          config.map {
+            case (key, value) if key.contains("password") || key.contains("token") => (key, "***")
+            case (key, value) => (key, value)
+          }
+        } else {
+          config
+        }
+        Connection(name, format, Some("data-source"), options - FORMAT)
+      case None =>
+        throw new IllegalArgumentException(s"Connection not found in application.conf: $name")
     }
   }
 
   private def getAllConnections(optConnectionGroupType: Option[String], masking: Boolean = true): GetConnectionsResponse = {
     LOGGER.debug(s"Getting all connection details, connection-group=${optConnectionGroupType.getOrElse("")}")
+    
+    // Get connections from files
     val connectionPath = Path.of(connectionSaveFolder)
     if (!connectionPath.toFile.exists()) connectionPath.toFile.mkdirs()
-    val connections = Files.list(connectionPath)
+    val fileConnections = Files.list(connectionPath)
       .iterator()
       .asScala
       .map(file => {
@@ -85,9 +126,38 @@ object ConnectionRepository {
       .filter(_.isSuccess)
       .map(_.get)
       .toList
+    
+    // Get connections from application.conf
+    val configConnections = getConnectionsFromConfig(masking)
+    
+    // Merge connections, with file connections taking priority (deduplicating by name)
+    val allConnections = (fileConnections ++ configConnections)
+      .groupBy(_.name)
+      .map(_._2.head) // Take first occurrence (file connection if exists, otherwise config)
+      .toList
       .filter(conn => optConnectionGroupType.forall(conn.groupType.contains))
       .sortBy(_.name)
-    GetConnectionsResponse(connections)
+    
+    GetConnectionsResponse(allConnections)
+  }
+  
+  /**
+   * Get all connections from application.conf
+   */
+  private def getConnectionsFromConfig(masking: Boolean = true): List[Connection] = {
+    val connectionConfigs = ConfigParser.connectionConfigsByName
+    connectionConfigs.map { case (name, config) =>
+      val format = config.getOrElse(FORMAT, "unknown")
+      val options = if (masking) {
+        config.map {
+          case (key, value) if key.contains("password") || key.contains("token") => (key, "***")
+          case (key, value) => (key, value)
+        }
+      } else {
+        config
+      }
+      Connection(name, format, Some("data-source"), options - FORMAT)
+    }.toList
   }
 
   private def removeConnection(connectionName: String): Boolean = {