Feat/mapping lookup observer

CLAUDE.md

@@ -38,7 +38,7 @@ Every plugin module plugs in through the `IgnifyrExtension` ServiceLoader SPI
 (`io.ignifyr.engine.spi`) — one implementation and one `META-INF/services` entry per jar. Its full
 contribution surface is `sourceConnectors`, `sinkProviders`, `terminologyServiceProviders`,
 `identityServiceProviders`, `cliCommands`, `sourceFailureDescriptors`, `schemaInferrers`,
-`streamingProvider`, `schedulerProvider`, `sparkConfContributions`, `extraCapabilities`, plus
+`streamingProvider`, `schedulerProvider`, `lookupObservers`, `sparkConfContributions`, `extraCapabilities`, plus
 `initialize(config)` (scoped to `ignifyr.extensions.<id>`; it must **not** touch
 `IgnifyrConfig.sparkSession` — registry load runs while the session is being built). The engine's
 `ExtensionRegistry` indexes these by settings/binding **class**, fail-fast on duplicate keys
@@ -47,6 +47,14 @@ connector, sink, or format directly. `ExtensionRegistry.init()` force-materializ
 can reject its input, **including `streaming` and `scheduler`**, so a second copy of a capability
 module fails at startup rather than at first job launch.
 
+`lookupObservers` (`MappingLookupObserver`, any number) is the coverage-monitoring hook: the engine
+reports every `mpp:getConcept` / `mpp:convertAndReturnQuantity` lookup and every terminology `translate` /
+`lookup` (via `ObservingTerminologyService`) as a `LookupEvent` to the executor-side `LookupRecorder` obtained
+in `FhirMappingJobManager.executeTask`, and calls `onChunkCompleted` from `SinkHandler.writeMappingResult`.
+It records and logs nothing itself — with no observer installed the recorder is `LookupRecorder.NoOp`, and
+previews/test runs (no execution id) are never observed. The Enterprise `ignifyr-observability` module is the
+implementation (aggregated `MAPPING_COVERAGE` events + a Kibana dashboard).
+
 `sparkConfContributions` is merged in three layers, lowest first: engine defaults → module
 contributions → the user's `spark { }` block. The user's block wins for a single-valued key, but for a
 key Spark parses as a comma-separated registration list (`IgnifyrConfig.additiveSparkConfKeys` —

ignifyr-engine/src/main/scala/io/ignifyr/engine/cli/command/ListPlugins.scala

@@ -47,6 +47,7 @@ class ListPlugins extends Command {
       line("schema inferrers", ext.schemaInferrers)(_.settingsClass.getSimpleName),
       line("CLI commands", ext.cliCommands.flatMap(c => c.name +: c.aliases).sorted)(identity),
       line("source failure descriptors", ext.sourceFailureDescriptors)(_.getClass.getSimpleName),
+      line("lookup observers", ext.lookupObservers)(_.getClass.getSimpleName),
       ext.streamingProvider.map(p => s"streaming provider: ${p.getClass.getSimpleName}"),
       ext.schedulerProvider.map(p => s"scheduler provider: ${p.getClass.getSimpleName}"),
       line("spark conf keys", ext.sparkConfContributions.keys.toSeq.sorted)(identity)

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

@@ -4,6 +4,7 @@ import com.typesafe.scalalogging.Logger
 import io.ignifyr.engine.execution.log.ExecutionLogger
 import io.ignifyr.engine.execution.processing.ErroneousRecordWriter
 import io.ignifyr.engine.model._
+import io.ignifyr.engine.spi.ExtensionRegistry
 import org.apache.spark.sql.{Dataset, SparkSession}
 import org.apache.spark.util.CollectionAccumulator
 
@@ -55,6 +56,8 @@ object SinkHandler {
       mappingErrors,
       invalidInputs
     )
+    // All actions on the mapped chunk are done; let the lookup observers publish what they recorded for it
+    ExtensionRegistry.notifyLookupChunkCompleted(mappingJobExecution, mappingTaskName)
     ErroneousRecordWriter.saveErroneousRecords(
       spark,
       mappingJobExecution,

ignifyr-engine/src/main/scala/io/ignifyr/engine/mapping/FhirMappingService.scala

@@ -6,9 +6,10 @@ import io.onfhir.path.{FhirPathEvaluator, IFhirPathFunctionLibraryFactory}
 import io.onfhir.template.FhirTemplateExpressionHandler
 import io.onfhir.definitions.common.model.Json4sSupport.formats
 import io.ignifyr.engine.mapping.fhirPath.FhirMappingFunctionsFactory
-import io.ignifyr.engine.mapping.service.IntegratedServiceFactory
+import io.ignifyr.engine.mapping.service.{IntegratedServiceFactory, ObservingTerminologyService}
 import io.ignifyr.engine.model._
 import io.ignifyr.engine.model.exception.FhirMappingException
+import io.ignifyr.engine.spi.LookupRecorder
 import org.json4s.JsonAST.{JArray, JNull, JObject, JValue}
 
 import io.ignifyr.engine.Execution.actorSystem.dispatcher
@@ -27,6 +28,8 @@ import scala.concurrent.Future
  * @param identityServiceSettings    Settings for identity service to use within mappings (e.g. resolveIdentifier)
  * @param functionLibraries          External function libraries containing functions to use in FHIRPath expressions
  * @param projectId                  Project identifier associated with the mapping job
+ * @param lookupRecorder             Recorder of the terminology / concept-map / unit-conversion lookups the mapping
+ *                                   performs (for coverage monitoring); records nothing by default
  *                                   (if true, mapped FHIR resources are grouped by input row in the FhirMappingResult)
  */
 class FhirMappingService(
@@ -39,11 +42,16 @@ class FhirMappingService(
     terminologyServiceSettings: Option[TerminologyServiceSettings],
     identityServiceSettings: Option[IdentityServiceSettings],
     functionLibraries: Map[String, IFhirPathFunctionLibraryFactory],
-    val projectId: Option[String]
+    val projectId: Option[String],
+    lookupRecorder: LookupRecorder = LookupRecorder.NoOp
 ) extends IFhirMappingService {
 
   lazy val terminologyService =
-    terminologyServiceSettings.map(setting => IntegratedServiceFactory.createTerminologyService(setting))
+    terminologyServiceSettings
+      .map(setting => IntegratedServiceFactory.createTerminologyService(setting))
+      .map(service =>
+        if (lookupRecorder eq LookupRecorder.NoOp) service else new ObservingTerminologyService(service, lookupRecorder)
+      )
   lazy val identityService =
     identityServiceSettings.map(setting => IntegratedServiceFactory.createIdentityService(setting))
 
@@ -57,7 +65,8 @@ class FhirMappingService(
         .map(c => c._1 -> c._2.toContextObject), // Provide the static contexts
       functionLibraries + // Default libraries
         ("mpp" -> new FhirMappingFunctionsFactory(
-          context.filterNot(_._2.isInstanceOf[ConfigurationContext])
+          context.filterNot(_._2.isInstanceOf[ConfigurationContext]),
+          lookupRecorder
         )), // Add our mapping function library,
       terminologyService,
       identityService,

ignifyr-engine/src/main/scala/io/ignifyr/engine/mapping/context/MappingContextLoader.scala

@@ -128,7 +128,11 @@ class MappingContextLoader extends IMappingContextLoader {
           case _ => Map.empty
         }
 
-      ConceptMapContext(concepts = concepts, conversionFunctions = conversionFunctions)
+      ConceptMapContext(
+        concepts = concepts,
+        conversionFunctions = conversionFunctions,
+        name = Some(new File(filePath).getName)
+      )
     }
   }
 

ignifyr-engine/src/main/scala/io/ignifyr/engine/mapping/fhirPath/FhirPathMappingFunctions.scala

@@ -11,6 +11,7 @@ import io.onfhir.path.annotation.{
 }
 import io.onfhir.path.grammar.FhirPathExprParser.ExpressionContext
 import io.ignifyr.engine.model.{ConceptMapContext, FhirMappingContext, UnitConversionContext}
+import io.ignifyr.engine.spi.{LookupEvent, LookupRecorder, LookupTypes}
 import io.ignifyr.engine.util.FhirMappingUtility
 import org.json4s.{JObject, JString}
 
@@ -20,11 +21,13 @@ import org.json4s.{JObject, JString}
  * @param context        FHIR Path context
  * @param current        Current result to apply the function on
  * @param mappingContext Specific mapping context
+ * @param lookupRecorder Recorder of the concept-map and unit-conversion lookups performed (for coverage monitoring)
  */
 class FhirPathMappingFunctions(
     context: FhirPathEnvironment,
     current: Seq[FhirPathResult],
-    mappingContext: Map[String, FhirMappingContext]
+    mappingContext: Map[String, FhirMappingContext],
+    lookupRecorder: LookupRecorder = LookupRecorder.NoOp
 ) extends AbstractFhirPathFunctionLibrary
     with Serializable {
 
@@ -384,17 +387,50 @@ class FhirPathMappingFunctions(
     evaluator.visit(keyExpr) match {
       case Nil => Seq.empty
       case Seq(FhirPathString(conceptCode)) =>
-        conceptMapContext.concepts
-          .get(conceptCode)
-          .map(mws => mws.map(res => FhirPathComplex(JObject(res.toList.map(i => i._1 -> JString(i._2))))))
-          .getOrElse(Seq.empty)
+        val matchedConcepts = conceptMapContext.concepts.getOrElse(conceptCode, Seq.empty)
+        recordConceptLookup(
+          conceptMapContext.name.getOrElse(mapName),
+          conceptCode,
+          matchedConcepts.flatMap(_.get("target_code")).find(_.nonEmpty),
+          matchedConcepts.nonEmpty
+        )
+        matchedConcepts.map(res => FhirPathComplex(JObject(res.toList.map(i => i._1 -> JString(i._2)))))
       case _ =>
         throw new FhirPathException(
           s"Invalid function call 'getConcept', given expression for keyExpr:${keyExpr.getText} for the concept code should return a string value!"
         )
     }
   }
 
+  /**
+   * Record a concept-map lookup (a concept map has no code systems, so only the codes are recorded)
+   *
+   * @param mapName    Name of the concept map context: the file it is loaded from, or its alias in the mapping
+   *                   if the file is unknown
+   * @param sourceCode The looked up code
+   * @param targetCode The matched value, if any
+   * @param success    Whether a match was found
+   * @param column     The column asked for; None when the whole concept is asked for
+   */
+  private def recordConceptLookup(
+      mapName: String,
+      sourceCode: String,
+      targetCode: Option[String],
+      success: Boolean,
+      column: Option[String] = None
+  ): Unit =
+    lookupRecorder.record(
+      LookupEvent(
+        lookupType = LookupTypes.CONCEPT,
+        operation = "getConcept",
+        conceptMap = Some(mapName),
+        sourceCode = Some(sourceCode),
+        targetCode = targetCode,
+        conceptColumn = column,
+        success = success
+      )
+    )
+
   /**
    * Load the concept map
    *
@@ -490,19 +526,23 @@ class FhirPathMappingFunctions(
       )
     }
     // If conceptCode returns empty, also return empty, if there is no such key or target column is null also return empty
-    val result: Seq[FhirPathResult] =
-      conceptCodeResult.headOption
-        .map(_.asInstanceOf[FhirPathString].s)
-        .flatMap { conceptCode =>
-          conceptMapContext.concepts.get(conceptCode).map { conceptMapEntries =>
-            conceptMapEntries
-              .flatMap(_.get(targetField))
-              .filter(_.nonEmpty)
-              .map(mappedValue => FhirPathString(mappedValue))
-          }
-        }
-        .getOrElse(Seq.empty)
-    result
+    conceptCodeResult.headOption.map(_.asInstanceOf[FhirPathString].s) match {
+      case None => Seq.empty
+      case Some(conceptCode) =>
+        val mappedValues =
+          conceptMapContext.concepts
+            .getOrElse(conceptCode, Seq.empty)
+            .flatMap(_.get(targetField))
+            .filter(_.nonEmpty)
+        recordConceptLookup(
+          conceptMapContext.name.getOrElse(mapName),
+          conceptCode,
+          mappedValues.headOption,
+          mappedValues.nonEmpty,
+          Some(targetField)
+        )
+        mappedValues.map(mappedValue => FhirPathString(mappedValue))
+    }
   }
 
   /**
@@ -575,7 +615,7 @@ class FhirPathMappingFunctions(
       try {
         mappingContext(mapName) match {
           case u: UnitConversionContext => u
-          case c: ConceptMapContext => UnitConversionContext(c.conversionFunctions)
+          case c: ConceptMapContext => UnitConversionContext(c.conversionFunctions, c.name)
           case _ => throw new Exception()
         }
       } catch {
@@ -586,7 +626,7 @@ class FhirPathMappingFunctions(
       }
 
     val codeResult = new FhirPathExpressionEvaluator(context, current).visit(keyExpr)
-    if (codeResult.length > 1 || !codeResult.head.isInstanceOf[FhirPathString]) {
+    if (codeResult.length > 1 || !codeResult.forall(_.isInstanceOf[FhirPathString])) {
       throw new FhirPathException(
         s"Invalid function call 'convertAndReturnQuantity', given expression for keyExpr:${keyExpr.getText} for the source code should return a string value!"
       )
@@ -597,7 +637,7 @@ class FhirPathMappingFunctions(
       FhirPathEvaluator.parse("utl:parseFhirQuantityExpression($this)")
     )
     valueResult = valueAndComparator.headOption.toSeq
-    if (valueResult.length > 1 || !valueResult.head.isInstanceOf[FhirPathNumber]) {
+    if (valueResult.length > 1 || !valueResult.forall(_.isInstanceOf[FhirPathNumber])) {
       throw new FhirPathException(
         s"Invalid function call 'convertAndReturnQuantity', given expression for valueExpr:${valueExpr.getText} for the value should return a numeric value!"
       )
@@ -616,37 +656,48 @@ class FhirPathMappingFunctions(
       Nil
     else {
       val code = codeResult.head.asInstanceOf[FhirPathString].s
-      unitConversionContext.conversionFunctions
-        .get(code -> unit)
-        .map { case (targetUnit, conversionFunction) =>
-          val conversionFunctionExpressionContext = FhirPathEvaluator.parse(conversionFunction)
-          val functionResult =
-            new FhirPathExpressionEvaluator(context, valueResult).visit(conversionFunctionExpressionContext)
-          if (functionResult.length != 1 || !functionResult.head.isInstanceOf[FhirPathNumber]) {
-            throw new FhirPathException(
-              s"Invalid FHIR expression in the unit conversion context! The FHIR path expression:${conversionFunction} should evaluate to a single numeric value!"
-            )
-          }
-          FhirPathComplex(
-            JObject(
-              List(
-                "value" -> functionResult.head.toJson,
-                "system" -> JString("http://unitsofmeasure.org"),
-                "unit" -> JString(targetUnit),
-                "code" -> JString(targetUnit)
-              ) ++
-                comparator.map(c => "comparator" -> JString(c)).toList
-            )
+      val conversion = unitConversionContext.conversionFunctions.get(code -> unit)
+      lookupRecorder.record(
+        LookupEvent(
+          lookupType = LookupTypes.UNIT,
+          operation = "convertAndReturnQuantity",
+          conceptMap = Some(unitConversionContext.name.getOrElse(mapName)),
+          sourceCode = Some(code),
+          sourceUnit = Some(unit),
+          targetUnit = conversion.map(_._1),
+          success = conversion.isDefined
+        )
+      )
+      conversion.map { case (targetUnit, conversionFunction) =>
+        val conversionFunctionExpressionContext = FhirPathEvaluator.parse(conversionFunction)
+        val functionResult =
+          new FhirPathExpressionEvaluator(context, valueResult).visit(conversionFunctionExpressionContext)
+        if (functionResult.length != 1 || !functionResult.head.isInstanceOf[FhirPathNumber]) {
+          throw new FhirPathException(
+            s"Invalid FHIR expression in the unit conversion context! The FHIR path expression:${conversionFunction} should evaluate to a single numeric value!"
           )
         }
-        .toSeq
+        FhirPathComplex(
+          JObject(
+            List(
+              "value" -> functionResult.head.toJson,
+              "system" -> JString("http://unitsofmeasure.org"),
+              "unit" -> JString(targetUnit),
+              "code" -> JString(targetUnit)
+            ) ++
+              comparator.map(c => "comparator" -> JString(c)).toList
+          )
+        )
+      }.toSeq
     }
   }
 }
 
-class FhirMappingFunctionsFactory(mappingContext: Map[String, FhirMappingContext])
-    extends IFhirPathFunctionLibraryFactory
+class FhirMappingFunctionsFactory(
+    mappingContext: Map[String, FhirMappingContext],
+    lookupRecorder: LookupRecorder = LookupRecorder.NoOp
+) extends IFhirPathFunctionLibraryFactory
     with Serializable {
   override def getLibrary(context: FhirPathEnvironment, current: Seq[FhirPathResult]): AbstractFhirPathFunctionLibrary =
-    new FhirPathMappingFunctions(context, current, mappingContext)
+    new FhirPathMappingFunctions(context, current, mappingContext, lookupRecorder)
 }

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

@@ -12,7 +12,13 @@ import io.ignifyr.engine.mapping.schema.IFhirSchemaLoader
 import io.ignifyr.engine.model._
 import io.ignifyr.engine.model.exception.{FhirMappingException, FhirMappingJobStoppedException}
 import io.ignifyr.engine.repository.mapping.IFhirMappingRepository
-import io.ignifyr.engine.spi.{ExtensionRegistry, MappingTaskPipeline, MissingCapabilityException}
+import io.ignifyr.engine.spi.{
+  ExtensionRegistry,
+  LookupRecorder,
+  MappingLookupScope,
+  MappingTaskPipeline,
+  MissingCapabilityException
+}
 import org.apache.spark.SparkThrowable
 import org.apache.spark.sql.functions.{collect_list, struct, udf}
 import org.apache.spark.sql.streaming.StreamingQuery
@@ -598,6 +604,12 @@ class FhirMappingJobManager(
       .map(loadedContextMap => {
         // Get configuration context
         val configurationContext = mainSourceSettings.toConfigurationContext
+        // Recorder of the lookups the mapping performs, only for executions (not for previews or test runs)
+        val lookupRecorder = executionId
+          .map(id =>
+            ExtensionRegistry.lookupRecorderFor(spark, MappingLookupScope(jobId, projectId, id, mappingTaskName))
+          )
+          .getOrElse(LookupRecorder.NoOp)
         // Construct the mapping service
         val fhirMappingService = new FhirMappingService(
           jobId,
@@ -609,7 +621,8 @@ class FhirMappingJobManager(
           terminologyServiceSettings,
           identityServiceSettings,
           functionLibraries,
-          projectId
+          projectId,
+          lookupRecorder
         )
         MappingTaskExecutor.executeMapping(spark, df, fhirMappingService, executionId)
       })