Skip Write Execution

README.md

@@ -194,6 +194,7 @@ When started as a standalone tool, the engine can run in two modes based on argu
 - `run`: Runs a configured mapping-job as a batch process and shuts down. `run` command accepts the following parameters:
   - `--job`: The path to the mapping-job to be executed. If provided, overrides the path provided to the JVM as the configuration parameter.
   - `--db` (alias `--db-path`): The path to the database folder that is used for scheduled jobs. If provided, overrides the path provided to the JVM as the configuration parameter.
+  - `--skip-write` (no value): Runs the mappings and logs their results, but writes nothing to the sink, does not validate the sink and does not archive the source files. Use it to test a job. Only batch jobs support it; a streaming or scheduled job is rejected.
 
   The mappings and schemas folders are not command-line options; they are always read from the
   [configuration](#configurations) (`ignifyr.mappings.repository.folder-path` and

ignifyr-engine/src/main/scala/io/ignifyr/engine/Boot.scala

@@ -38,7 +38,12 @@ object Boot extends App {
           if (options.contains("db-path")) options("db-path").asInstanceOf[String]
           else IgnifyrConfig.engineConfig.ignifyrDbFolderPath
 
-        CommandLineInterface.runJob(ignifyrEngine, mappingJobFilePath, ignifyrDbFolderPath)
+        CommandLineInterface.runJob(
+          ignifyrEngine,
+          mappingJobFilePath,
+          ignifyrDbFolderPath,
+          skipWrite = options.contains("skip-write")
+        )
 
       // Diagnostic: list the installed extension modules and everything they contribute. Reads only
       // the ServiceLoader-backed registry, so it deliberately does NOT build the engine (no workspace

ignifyr-engine/src/main/scala/io/ignifyr/engine/cli/CommandLineInterface.scala

@@ -84,14 +84,21 @@ object CommandLineInterface {
    *
    * @param ignifyrEngine
    * @param mappingJobFilePath
+   * @param skipWrite Run the mappings without writing their results to the sink (batch jobs only)
    */
-  def runJob(ignifyrEngine: IgnifyrEngine, mappingJobFilePath: Option[String], ignifyrDbFolderPath: String): Unit = {
+  def runJob(
+      ignifyrEngine: IgnifyrEngine,
+      mappingJobFilePath: Option[String],
+      ignifyrDbFolderPath: String,
+      skipWrite: Boolean = false
+  ): Unit = {
     if (mappingJobFilePath.isEmpty) {
       println("There are no jobs to run. Exiting...")
       System.exit(1)
     }
     val mappingJob = FhirMappingJobFormatter.readMappingJobFromFile(mappingJobFilePath.get)
-    val mappingJobExecution = FhirMappingJobExecution(job = mappingJob, mappingTasks = mappingJob.mappings)
+    val mappingJobExecution =
+      FhirMappingJobExecution(job = mappingJob, mappingTasks = mappingJob.mappings, skipWrite = skipWrite)
     new MappingJobLauncher(ignifyrEngine).launch(mappingJob, mappingJobExecution, ignifyrDbFolderPath) match {
       case MappingJobLaunch.Batch(completion) =>
         Await.result(completion, Duration.Inf)
@@ -117,6 +124,9 @@ object CommandLineInterface {
       case Nil => map
       case ("--db" | "--db-path") :: value :: tail =>
         nextArg(map ++ Map("db-path" -> value), tail)
+      case "--skip-write" :: tail =>
+        // A valueless boolean flag; it must match before the generic pair below swallows the next token.
+        nextArg(map ++ Map("skip-write" -> true), tail)
       case flag :: value :: tail if flag.startsWith("--") =>
         // Generic `--flag value` pair; command providers translate these into positional args.
         nextArg(map ++ Map(flag.stripPrefix("--") -> value), tail)

ignifyr-engine/src/main/scala/io/ignifyr/engine/data/write/SinkHandler.scala

@@ -39,8 +39,14 @@ object SinkHandler {
     val fhirWriteProblemsAccum: CollectionAccumulator[FhirMappingResult] =
       spark.sparkContext.collectionAccumulator[FhirMappingResult](accumName)
     fhirWriteProblemsAccum.reset()
-    // Write the FHIR resources
-    resourceWriter.write(spark, mappedResults, fhirWriteProblemsAccum)
+    // Write the FHIR resources, unless the execution skips writing (the accumulator then stays empty, so the mapped
+    // resources are logged as numOfFhirResources with no failed writes)
+    if (mappingJobExecution.isWriteSkipped)
+      logger.info(
+        s"Skipping the write of the mapped resources for mappingTask '$mappingTaskName' of execution '${mappingJobExecution.id}'."
+      )
+    else
+      resourceWriter.write(spark, mappedResults, fhirWriteProblemsAccum)
     logMappingJobResult(
       mappingJobExecution,
       mappingTaskName,

ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/MappingJobLauncher.scala

@@ -69,13 +69,23 @@ class MappingJobLauncher(ignifyrEngine: IgnifyrEngine)(implicit ec: ExecutionCon
    * @param ignifyrDbFolderPath  Ignifyr database folder (scheduling providers keep last-sync state there).
    * @param clearCheckpoints     For streaming jobs, reset archiving offsets and delete the Spark
    *                             checkpoint directories so the streams start from scratch.
+   * @throws IllegalArgumentException if the execution skips writing but the job is streaming or scheduled: skipping
+   *                                  the write is supported only for batch executions, since a stream would still
+   *                                  advance its checkpoints and a schedule its last-sync time.
    */
   def launch(
       mappingJob: FhirMappingJob,
       mappingJobExecution: FhirMappingJobExecution,
       ignifyrDbFolderPath: String = IgnifyrConfig.engineConfig.ignifyrDbFolderPath,
       clearCheckpoints: Boolean = false
   ): MappingJobLaunch = {
+    if (
+      mappingJobExecution.isWriteSkipped &&
+      (mappingJob.sourceSettings.exists(_._2.asStream) || mappingJob.schedulingSettings.nonEmpty)
+    )
+      throw new IllegalArgumentException(
+        s"Skipping the write is supported only for batch executions, but job '${mappingJob.id}' is streaming or scheduled."
+      )
     if (mappingJob.sourceSettings.exists(_._2.asStream)) {
       if (clearCheckpoints) clearCheckpointDirectories(mappingJobExecution)
       val queryRegistrations = jobManager

ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/log/ExecutionLogger.scala

@@ -187,6 +187,7 @@ object ExecutionLogger {
     markerMap.put("projectId", mappingJobExecution.projectId)
     markerMap.put("executionId", mappingJobExecution.id)
     markerMap.put("mappingTaskName", mappingTaskName)
+    markerMap.put("isWriteSkipped", mappingJobExecution.isWriteSkipped)
     // log the chunk progress for batch jobs
     if (!mappingJobExecution.isStreamingJob) {
       markerMap.put("chunkProgress", s"0 / $numOfChunks")
@@ -238,6 +239,7 @@ object ExecutionLogger {
     markerMap.put("projectId", mappingJobExecution.projectId)
     markerMap.put("executionId", mappingJobExecution.id)
     markerMap.put("mappingTaskName", mappingTaskName)
+    markerMap.put("isWriteSkipped", mappingJobExecution.isWriteSkipped)
     // log the batch progress for batch jobs
     if (!mappingJobExecution.isStreamingJob) {
       markerMap.put("chunkProgress", s"0 / 1") // chunk size is not determined yet

ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/processing/FileStreamInputArchiver.scala

@@ -102,7 +102,12 @@ object FileStreamInputArchiver {
     // That means archiving works in best-effort mode.
     try {
       val archiveMode: ArchiveModes = execution.archiveMode
-      if (archiveMode != ArchiveModes.OFF) {
+      // An execution that skips writing must leave its input files in place: it wrote nothing, so they are not consumed
+      if (execution.isWriteSkipped && archiveMode != ArchiveModes.OFF)
+        logger.info(
+          s"Skipping archiving for job: ${execution.jobId}, execution: ${execution.id}, since its write is skipped"
+        )
+      else if (archiveMode != ArchiveModes.OFF) {
         // get data folder path from data source settings
         val dataFolderPath = FileUtils.getPath(execution.fileSystemSourceDataFolderPath.get).toString
 

ignifyr-engine/src/main/scala/io/ignifyr/engine/mapping/job/FhirMappingJobManager.scala

@@ -90,7 +90,8 @@ class FhirMappingJobManager(
       timeRange: Option[(LocalDateTime, LocalDateTime)] = None
   ): Future[Unit] = {
     val sinkWriter = SinkWriterFactory.apply(sinkSettings)
-    sinkWriter.validate()
+    // The sink is not touched at all when the execution skips writing
+    if (!mappingJobExecution.isWriteSkipped) sinkWriter.validate()
     mappingJobExecution.mappingTasks.foldLeft(Future((): Unit)) { (f, task) => // Initial empty Future
       f.flatMap { _ => // Execute the Futures in the Sequence consecutively (not in parallel)
         // log the start of the FHIR mapping task execution
@@ -212,7 +213,8 @@ class FhirMappingJobManager(
       identityServiceSettings: Option[IdentityServiceSettings] = None
   ): Future[Unit] = {
     val sinkWriter = SinkWriterFactory.apply(sinkSettings)
-    sinkWriter.validate()
+    // The sink is not touched at all when the execution skips writing
+    if (!mappingJobExecution.isWriteSkipped) sinkWriter.validate()
 
     readSourceAndExecuteTask(
       mappingJobExecution.jobId,

ignifyr-engine/src/main/scala/io/ignifyr/engine/model/FhirMappingJobExecution.scala

@@ -21,6 +21,9 @@ import java.util.regex.Pattern
  * @param archiveMode                    Archive mode of execution
  * @param saveErroneousRecords           Whether to save erroneous records or not
  * @param isScheduledJob                 Whether the execution is scheduled or not
+ * @param isWriteSkipped                 Whether writing the mapped resources to the sink is skipped (a dry run: the
+ *                                       mappings run and their results are logged, but nothing reaches the sink and
+ *                                       the source files are not archived). Supported only for batch executions.
  */
 case class FhirMappingJobExecution(
     id: String,
@@ -32,7 +35,8 @@ case class FhirMappingJobExecution(
     fileSystemSourceDataFolderPath: Option[String],
     archiveMode: ArchiveModes,
     saveErroneousRecords: Boolean,
-    isScheduledJob: Boolean
+    isScheduledJob: Boolean,
+    isWriteSkipped: Boolean
 ) {
 
   /**
@@ -144,14 +148,16 @@ object FhirMappingJobExecution {
    * @param job                            FHIR mapping job that includes this execution.
    * @param mappingTasks                   List of mapping tasks to be executed (as a subset of the mapping tasks defined in the job)
    * @param jobGroupIdOrStreamingQuery     Keeps Spark job group id for batch jobs and StreamingQuery for streaming jobs
+   * @param skipWrite                      Whether to skip writing the mapped resources to the sink (see [[FhirMappingJobExecution.isWriteSkipped]])
    * @return
    */
   def apply(
       id: String = UUID.randomUUID().toString,
       projectId: String = "",
       job: FhirMappingJob,
       mappingTasks: Seq[FhirMappingTask] = Seq.empty,
-      jobGroupIdOrStreamingQuery: Option[Either[String, collection.mutable.Map[String, StreamingQuery]]] = None
+      jobGroupIdOrStreamingQuery: Option[Either[String, collection.mutable.Map[String, StreamingQuery]]] = None,
+      skipWrite: Boolean = false
   ): FhirMappingJobExecution = {
 
     // Configure properties related to the source settings of the job
@@ -186,7 +192,8 @@ object FhirMappingJobExecution {
       fileSystemSourceDataFolderPath,
       archiveMode,
       saveErroneousRecords,
-      isScheduledJob
+      isScheduledJob,
+      skipWrite
     )
   }
 }

ignifyr-engine/src/main/scala/io/ignifyr/engine/model/FhirMappingJobResult.scala

@@ -66,7 +66,8 @@ case class FhirMappingJobResult(
         s"\t# of Invalid Rows: \t$numOfInvalids\n" +
         s"\t# of Not Mapped: \t$numOfNotMapped\n" +
         s"\t# of Failed writes:\t$numOfFailedWrites\n" +
-        s"\t# of Written FHIR resources:\t$numOfFhirResources"
+        (if (mappingJobExecution.isWriteSkipped) s"\t# of FHIR resources (write skipped):\t$numOfFhirResources"
+         else s"\t# of Written FHIR resources:\t$numOfFhirResources")
     }
     message
   }
@@ -99,6 +100,7 @@ case class FhirMappingJobResult(
     markerMap.put("eventId", eventId)
     markerMap.put("isStreamingJob", mappingJobExecution.isStreamingJob)
     markerMap.put("isScheduledJob", mappingJobExecution.isScheduledJob)
+    markerMap.put("isWriteSkipped", mappingJobExecution.isWriteSkipped)
     // log the chunk progress for batch jobs
     if (!mappingJobExecution.isStreamingJob) {
       markerMap.put("chunkProgress", s"$completedNumOfChunks / $totalNumOfChunks")

ignifyr-engine/src/test/scala/io/ignifyr/test/engine/cli/CommandLineInterfaceTest.scala

@@ -64,4 +64,18 @@ class CommandLineInterfaceTest extends AnyFlatSpec with Matchers {
   it should "let a later flag override an earlier one with the same name" in {
     parse("run", "--job", "first.json", "--job", "second.json") should contain("job" -> "second.json")
   }
+
+  // --skip-write takes no value: read as a generic pair it would swallow the next token (here `--job`),
+  // and the job path would be lost.
+  it should "read --skip-write as a valueless flag anywhere on the line" in {
+    parse("run", "--skip-write", "--job", "jobs/patient.json") shouldBe
+      Map("command" -> "run", "skip-write" -> true, "job" -> "jobs/patient.json")
+    parse("run", "--job", "jobs/patient.json", "--skip-write") shouldBe
+      Map("command" -> "run", "skip-write" -> true, "job" -> "jobs/patient.json")
+    parse("--skip-write", "run") shouldBe Map("command" -> "run", "skip-write" -> true)
+  }
+
+  it should "not set skip-write unless it is given" in {
+    parse("run", "--job", "jobs/patient.json") should not contain key("skip-write")
+  }
 }

ignifyr-engine/src/test/scala/io/ignifyr/test/engine/data/SinkHandlerTest.scala

@@ -25,11 +25,15 @@ class SinkHandlerTest extends AnyFlatSpec with Matchers {
   /** Captures the dataset it is handed instead of writing it anywhere. */
   private class CapturingWriter extends BaseSinkWriter(FileSystemSinkSettings("./out", SinkContentTypes.NDJSON)) {
     var written: Seq[FhirMappingResult] = Seq.empty
+    var writeCalls: Int = 0
     override def write(
         spark: SparkSession,
         df: Dataset[FhirMappingResult],
         problemsAccumulator: CollectionAccumulator[FhirMappingResult]
-    ): Unit = written = df.collect().toSeq
+    ): Unit = {
+      writeCalls += 1
+      written = df.collect().toSeq
+    }
     override def validate(): Unit = ()
   }
 
@@ -104,6 +108,24 @@ class SinkHandlerTest extends AnyFlatSpec with Matchers {
   }
 
   it should "call the writer even when there is nothing to write" in {
-    write() shouldBe empty
+    import sparkSession.implicits._
+    val writer = new CapturingWriter
+    SinkHandler.writeMappingResult(sparkSession, execution, "task-1", Seq.empty[FhirMappingResult].toDS(), writer)
+    writer.writeCalls shouldBe 1
+  }
+
+  // An execution that skips writing still maps and logs, but its mapped resources never reach the writer.
+  it should "not call the writer when the execution skips writing" in {
+    import sparkSession.implicits._
+    val writer = new CapturingWriter
+    SinkHandler.writeMappingResult(
+      sparkSession,
+      execution.copy(isWriteSkipped = true),
+      "task-1",
+      Seq(result("row-1", mappedResource = Some("""{"resourceType":"Patient"}"""))).toDS(),
+      writer
+    )
+    writer.writeCalls shouldBe 0
+    writer.written shouldBe empty
   }
 }

ignifyr-engine/src/test/scala/io/ignifyr/test/engine/execution/FileStreamInputArchiverTest.scala

@@ -181,6 +181,24 @@ class FileStreamInputArchiverTest extends AnyFlatSpec with Matchers {
     org.apache.commons.io.FileUtils.deleteDirectory(FileUtils.getPath(sourceFolderPath).toFile)
   }
 
+  "FileStreamInputArchiver" should "not apply deletion for a batch job that skips writing" in {
+
+    // Create a test input file
+    val inputFile = initializeInputFiles(sourceFolderPath, inputFilePath)
+
+    // Check whether input file exists
+    inputFile.exists() shouldBe true
+
+    // Call archiving function for an execution that wrote nothing
+    FileStreamInputArchiver.applyArchivingOnBatchJob(testExecutionWithDelete.copy(isWriteSkipped = true))
+
+    // Check whether input file remains
+    inputFile.exists() shouldBe true
+
+    // Clean test directories
+    org.apache.commons.io.FileUtils.deleteDirectory(FileUtils.getPath(sourceFolderPath).toFile)
+  }
+
   "FileStreamInputArchiver" should "get input files" in {
 
     val mappingTaskName = "mocked_mappingTask_name"

ignifyr-engine/src/test/scala/io/ignifyr/test/engine/execution/MappingJobLauncherSkipWriteTest.scala

@@ -0,0 +1,45 @@
+package io.ignifyr.test.engine.execution
+
+import io.ignifyr.engine.IgnifyrEngine
+import io.ignifyr.engine.execution.MappingJobLauncher
+import io.ignifyr.engine.model._
+import org.mockito.MockitoSugar._
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.concurrent.ExecutionContext.Implicits.global
+
+/**
+ * Skipping the write is supported only for batch executions: a streaming execution would still advance its
+ * checkpoints and a scheduled one its last-sync time, so a later real run would silently skip the data the
+ * skipped run consumed. The launcher, the single dispatch point for the CLI and the server, refuses both
+ * before anything starts.
+ */
+class MappingJobLauncherSkipWriteTest extends AnyFlatSpec with Matchers {
+
+  private val launcher = new MappingJobLauncher(mock[IgnifyrEngine])
+
+  private val sinkSettings = FileSystemSinkSettings("./out", SinkContentTypes.NDJSON)
+
+  private def job(asStream: Boolean = false, scheduling: Option[BaseSchedulingSettings] = None): FhirMappingJob =
+    FhirMappingJob(
+      id = "job-1",
+      sourceSettings = Map("_" -> FileSystemSourceSettings("test", "test", "data", asStream = asStream)),
+      sinkSettings = sinkSettings,
+      mappings = Seq.empty,
+      schedulingSettings = scheduling
+    )
+
+  private def skippedExecution(mappingJob: FhirMappingJob): FhirMappingJobExecution =
+    FhirMappingJobExecution(job = mappingJob, skipWrite = true)
+
+  "launch" should "refuse to skip the write of a streaming job" in {
+    val streamingJob = job(asStream = true)
+    an[IllegalArgumentException] should be thrownBy launcher.launch(streamingJob, skippedExecution(streamingJob))
+  }
+
+  it should "refuse to skip the write of a scheduled job" in {
+    val scheduledJob = job(scheduling = Some(SchedulingSettings("* * * * *")))
+    an[IllegalArgumentException] should be thrownBy launcher.launch(scheduledJob, skippedExecution(scheduledJob))
+  }
+}

ignifyr-engine/src/test/scala/io/ignifyr/test/engine/model/FhirMappingJobExecutionTest.scala

@@ -1,6 +1,11 @@
 package io.ignifyr.test.engine.model
 
-import io.ignifyr.engine.model.{FhirMappingJob, FhirMappingJobExecution, FhirRepositorySinkSettings}
+import io.ignifyr.engine.model.{
+  FhirMappingJob,
+  FhirMappingJobExecution,
+  FhirMappingJobResult,
+  FhirRepositorySinkSettings
+}
 import io.ignifyr.engine.util.SparkUtil
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
@@ -28,4 +33,24 @@ class FhirMappingJobExecutionTest extends AnyFlatSpec with Matchers {
     testExecution.getCommitDirectory(mappingTaskName) shouldBe
       SparkUtil.getCommitDirectoryPath(Paths.get(testExecution.getCheckpointDirectory(mappingTaskName)))
   }
+
+  "FhirMappingJobExecution" should "write its results unless skipping the write is asked for" in {
+    testExecution.isWriteSkipped shouldBe false
+    FhirMappingJobExecution(job = testJob, skipWrite = true).isWriteSkipped shouldBe true
+  }
+
+  // The Kibana dashboards read `isWriteSkipped` off every result event to tell a written execution from a skipped one
+  "FhirMappingJobResult" should "carry whether the write is skipped in its log marker" in {
+    def markerOf(execution: FhirMappingJobExecution) =
+      FhirMappingJobResult(execution, Some(mappingTaskName)).toMapMarker.getMap
+    markerOf(testExecution).get("isWriteSkipped") shouldBe false
+    markerOf(testExecution.copy(isWriteSkipped = true)).get("isWriteSkipped") shouldBe true
+  }
+
+  it should "not report the resources of a skipped write as written" in {
+    val result =
+      FhirMappingJobResult(testExecution.copy(isWriteSkipped = true), Some(mappingTaskName), chunkResult = false)
+    result.toString should include("write skipped")
+    result.toString should not include "Written"
+  }
 }