Transform task: added an optional limit parameter that restricts the transformation to the first N input entities (CMEM-7621)

silk-plugins/silk-persistent-caching/src/main/scala/org/silkframework/plugins/dataset/hierarchical/HierarchicalSink.scala

@@ -18,8 +18,13 @@ abstract class HierarchicalSink extends EntitySink {
   // Holds root entities
   private val rootEntities: SequentialEntityCache = SequentialEntityCache()
 
-  // Holds nested entities
-  private lazy val cache: HierarchicalEntityCache = HierarchicalEntityCache()
+  // Holds nested entities. Lazily created; `cacheUsed` tracks whether it was initialized so close() does
+  // not spin up the persistent store for flat outputs.
+  private var cacheUsed: Boolean = false
+  private lazy val cache: HierarchicalEntityCache = {
+    cacheUsed = true
+    HierarchicalEntityCache()
+  }
 
   // All properties for each table.
   private val tables: mutable.Buffer[TableSpec] = mutable.Buffer.empty
@@ -84,7 +89,9 @@ abstract class HierarchicalSink extends EntitySink {
         outputEntities(writeEntities)
       }
     } finally {
-      cache.close()
+      if(cacheUsed) {
+        cache.close()
+      }
       rootEntities.close()
     }
   }

silk-plugins/silk-persistent-caching/src/main/scala/org/silkframework/runtime/caching/PersistentSortedKeyValueStore.scala

@@ -425,9 +425,11 @@ object PersistentSortedKeyValueStore {
     file
   }
 
-  /** Directory where temporary databases are stored that will be removed on every start of the application. */
+  /** Directory for temporary databases, wiped recursively on first use ([[removeTempDirectories]]).
+    * Must stay exclusive to this store and not overlap a shared temp dir like `config.tempFilesDirectory`,
+    * otherwise the startup wipe would delete unrelated temp files. */
   def tempCacheDirectory: File = {
-    new File(cacheDirectory, "tmp")
+    new File(cacheDirectory, "kvstore-tmp")
   }
 
   private def removeTempDirectories(): Unit = {

silk-plugins/silk-persistent-caching/src/test/scala/org/silkframework/runtime/caching/PersistentSortedKeyValueStoreTest.scala

@@ -23,21 +23,33 @@ class PersistentSortedKeyValueStoreTest extends AnyFlatSpec with Matchers {
     // This check should work
     PersistentSortedKeyValueStore.check()
 
-    // Create a file where LMDB would place its database in order to break it
+    // The check should fail now
     val dir = Files.createTempDirectory("ldmbBootTest")
-    val dbBaseDir = dir.resolve("tmp")
-    dbBaseDir.toFile.mkdirs()
-    Files.createFile(dbBaseDir.resolve("bootTest"))
+    ConfigTestTrait.withConfig("caches.persistence.directory" -> Some(dir.toFile.getCanonicalPath)) {
+      // Plant a file where LMDB would place the "bootTest" database directory, in order to break it.
+      val dbBaseDir = PersistentSortedKeyValueStore.tempCacheDirectory
+      dbBaseDir.mkdirs()
+      Files.createFile(dbBaseDir.toPath.resolve("bootTest"))
 
-    // The check should fail now
-    ConfigTestTrait.withConfig(("caches.persistence.directory" -> Some(dir.toFile.getCanonicalPath))) {
       an [LmdbException] shouldBe thrownBy { PersistentSortedKeyValueStore.check().get }
     }
 
     // Cleanup
     dir.toFile.deleteRecursive()
   }
 
+  it should "store temporary databases in a dedicated directory, not a shared temp directory" in {
+    TestFileUtils.withTempDirectory { cacheDir =>
+      ConfigTestTrait.withConfig("caches.persistence.directory" -> Some(cacheDir.getCanonicalPath)) {
+        // The temp DB directory is wiped recursively on startup, so it must stay exclusive to the store and
+        // not coincide with a generic 'tmp' dir shared with FileUtils.tempDir (which would delete unrelated files).
+        val tempDbDir = PersistentSortedKeyValueStore.tempCacheDirectory.getCanonicalFile
+        tempDbDir.getParentFile mustBe cacheDir.getCanonicalFile
+        tempDbDir.getName must not be "tmp"
+      }
+    }
+  }
+
   it should "store and retrieve single string values to/from the store" in {
     withStore() { store =>
       for(value <- values) {

silk-plugins/silk-serialization-json/src/test/scala/org/silkframework/serialization/json/PluginSerializersTest.scala

@@ -38,7 +38,7 @@ class PluginSerializersTest extends AnyFlatSpec with Matchers {
     stringValue(js, "pluginId") mustBe "transform"
     stringValueOption(js, TASKTYPE) mustBe empty
     val properties = objectValue(js, PROPERTIES)
-    properties.keys mustBe Set("selection", "mappingRule", "output", "errorOutput", "targetVocabularies", "abortIfErrorsOccur")
+    properties.keys mustBe Set("selection", "mappingRule", "output", "errorOutput", "targetVocabularies", "abortIfErrorsOccur", "inputLimit")
     mustBeJsObject(properties.value("selection")) { operatorsParam =>
       stringValue(operatorsParam, "type") mustBe "object"
       stringValue(operatorsParam, "pluginId") mustBe "datasetSelectionParameter"

silk-rules/src/main/scala/org/silkframework/rule/TransformSpec.scala

@@ -11,7 +11,7 @@ import org.silkframework.rule.vocab.TargetVocabularyParameterEnum
 import org.silkframework.runtime.plugin.StringParameterType.{EnumerationType, StringIterableParameterType}
 import org.silkframework.runtime.plugin._
 import org.silkframework.runtime.plugin.annotations.{Param, Plugin}
-import org.silkframework.runtime.plugin.types.IdentifierOptionParameter
+import org.silkframework.runtime.plugin.types.{IdentifierOptionParameter, IntOptionParameter}
 import org.silkframework.runtime.resource.Resource
 import org.silkframework.runtime.serialization.XmlSerialization._
 import org.silkframework.runtime.serialization.{ReadContext, WriteContext, XmlFormat, XmlSerialization}
@@ -62,7 +62,12 @@ case class TransformSpec(@Param(label = "Input", value = "The source from which
                          targetVocabularies: TargetVocabularyParameter = TargetVocabularyCategory(TargetVocabularyParameterEnum.allInstalled),
                          @Param("If true, a validation error (such as a data type mismatch) will abort the execution. " +
                                 "If false, the execution will continue, adding a validation error to the execution report.")
-                         abortIfErrorsOccur: Boolean = false
+                         abortIfErrorsOccur: Boolean = false,
+                         @Param(label = "Input limit", value = "If set, only the first N input entities are transformed. " +
+                             "Note: for hierarchical mappings, the limit is applied to each level independently. This can lead to inconsistent results between " +
+                             "parent and child tables, because a kept parent entity may reference child entities that were left out (and vice versa). " +
+                             "For this reason, some datasets like XML or JSON might fail during writing.", advanced = true)
+                         inputLimit: IntOptionParameter = None
                         ) extends TaskSpec with AnyPlugin {
 
   /** Retrieves the root rules of this transform spec. */

silk-rules/src/main/scala/org/silkframework/rule/execution/ExecuteTransform.scala

@@ -20,11 +20,13 @@ class ExecuteTransform(task: Task[TransformSpec],
                        input: UserContext => DataSource,
                        output: UserContext => EntitySink,
                        errorOutput: UserContext => Option[EntitySink] = _ => None,
-                       pluginContext: UserContext => PluginContext,
-                       limit: Option[Int] = None) extends Activity[TransformReport] {
+                       pluginContext: UserContext => PluginContext) extends Activity[TransformReport] {
 
   private def transform = task.data
 
+  /** Optional limit on the number of input entities to transform, configured on the transform task. */
+  private def inputLimit: Option[Int] = transform.inputLimit
+
   require(transform.rules.count(_.target.isEmpty) <= 1, "Only one rule with empty target property (subject rule) allowed.")
 
   override val initialValue = Some(TransformReport(task))
@@ -87,22 +89,23 @@ class ExecuteTransform(task: Task[TransformSpec],
     errorEntitySink.foreach(_.openTable(rule.outputSchema.typeUri, rule.outputSchema.typedPaths.map(_.property.get) :+ ErrorOutputWriter.errorProperty, singleEntity))
 
     val entityTable = try {
-      dataSource.retrieve(rule.inputSchema)
+      // Push the input limit down to the data source (best-effort; not every source honors it).
+      dataSource.retrieve(rule.inputSchema, inputLimit)
     } catch {
       case NonFatal(ex) =>
         throw new RuntimeException("Failed to retrieve input entities from data source.", ex)
     }
-    val transformedEntities = new TransformedEntities(task, entityTable.entities, rule.transformRule.label(), rule.transformRuleExecution, rule.outputSchema,
+    // Enforce the input limit client-side as well, since the push-down is best-effort.
+    val inputEntities = inputLimit.map(entityTable.entities.take).getOrElse(entityTable.entities)
+    val transformedEntities = new TransformedEntities(task, inputEntities, rule.transformRule.label(), rule.transformRuleExecution, rule.outputSchema,
       isRequestedSchema = false, abortIfErrorsOccur = task.data.abortIfErrorsOccur, report = reportBuilder).iterator
-    var count = 0
     breakable {
       for (entity <- transformedEntities) {
         entitySink.writeEntity(entity.uri, entity.values)
         if(entity.hasFailed) {
           errorEntitySink.foreach(_.writeEntity(entity.uri, entity.values :+ Seq(entity.failure.get.message.getOrElse("Unknown error"))))
         }
-        count += 1
-        if (cancelled || limit.exists(_ <= count)) {
+        if (cancelled) {
           break()
         }
       }

silk-rules/src/main/scala/org/silkframework/rule/execution/local/LocalTransformSpecExecutor.scala

@@ -58,6 +58,8 @@ class LocalTransformSpecExecutor extends Executor[TransformSpec, LocalExecution]
     val rule = schemata.transformRule
     val ruleLabel = rule.label()
     val requestedOutputType: Option[Uri] = requestedOutputSchema.map(_.typeUri)
+    // Optional limit on the number of input entities to transform, configured on the transform task.
+    val inputLimit: Option[Int] = task.data.inputLimit
 
     requestedOutputType match {
       case Some(outputType) =>
@@ -67,14 +69,16 @@ class LocalTransformSpecExecutor extends Executor[TransformSpec, LocalExecution]
         val inputTables = flattenInputs(input).toBuffer
         val (requestedRuleLabel, requestedRules, inputTable) = findMappingRulesMatchingRequestedOutputSchema(rules, ruleLabel, outputType, inputTables)
         addInputErrorsToTransformReport(inputTable, report)
-        val transformedEntities = new TransformedEntities(task, inputTable.entities, requestedRuleLabel,
+        val inputEntities = inputLimit.map(inputTable.entities.take).getOrElse(inputTable.entities)
+        val transformedEntities = new TransformedEntities(task, inputEntities, requestedRuleLabel,
           rule.withChildren(requestedRules).execution(taskContext), activeOutputSchema,
           isRequestedSchema = true, abortIfErrorsOccur = task.data.abortIfErrorsOccur, report).iterator
         GenericEntityTable(transformedEntities, activeOutputSchema, task)
       case _ =>
         // Else execute the complete mapping
         addInputErrorsToTransformReport(input, report)
-        val transformedEntities = new TransformedEntities(task, input.entities, ruleLabel, rule.execution(taskContext), schemata.outputSchema,
+        val inputEntities = inputLimit.map(input.entities.take).getOrElse(input.entities)
+        val transformedEntities = new TransformedEntities(task, inputEntities, ruleLabel, rule.execution(taskContext), schemata.outputSchema,
           isRequestedSchema = false, abortIfErrorsOccur = task.data.abortIfErrorsOccur, report).iterator
         GenericEntityTable(transformedEntities, schemata.outputSchema, task)
     }

silk-rules/src/test/scala/org/silkframework/rule/execution/ExecuteTransformTest.scala

@@ -52,6 +52,31 @@ class ExecuteTransformTest extends AnyFlatSpec with Matchers with MockitoSugar {
     resultStats.ruleResults("prop2Transform").errorCount shouldBe 0
   }
 
+  it should "transform only the first N entities when an input limit is set" in {
+    val prop = "http://prop"
+    val outputMock = mock[EntitySink]
+    val entities = Seq(entity("a", prop), entity("b", prop), entity("c", prop))
+    val dataSourceMock = mock[DataSource]
+    when(dataSourceMock.retrieve(any(), any())(any())).thenReturn(GenericEntityTable(CloseableIterator(entities.iterator), entities.head.schema, null))
+    when(dataSourceMock.underlyingTask).thenReturn(PlainTask("inputTaskDummy", DatasetSpec(InternalDataset())))
+    val transform = TransformSpec(datasetSelection(), RootMappingRule(rules = MappingRules(mapping("propTransform", prop))), inputLimit = Some(2))
+    val execute = new ExecuteTransform(
+      PlainTask("transformTask", transform),
+      inputTask = _ => PlainTask("dummy", DatasetSpec(EmptyDataset)),
+      input = _ => dataSourceMock,
+      output = _ => outputMock,
+      pluginContext = _ => PluginContext.empty,
+    )
+    val contextMock = mock[ActivityContext[TransformReport]]
+    val executeTransformResultHolder = new ValueHolder[TransformReport](None)
+    when(contextMock.value).thenReturn(executeTransformResultHolder)
+    when(contextMock.status).thenReturn(mock[StatusHolder])
+    implicit val userContext: UserContext = UserContext.Empty
+    execute.run(contextMock)
+    // The input limit is enforced, so only the first two of the three source entities are transformed.
+    executeTransformResultHolder().entityCount shouldBe 2
+  }
+
   private def transformerWithExceptions(): Transformer = {
     new InlineTransformer {
       override def apply(values: Seq[Seq[String]]): Seq[String] = {

silk-workspace/src/main/scala/org/silkframework/workspace/activity/transform/ExecuteTransformFactory.scala

@@ -5,8 +5,7 @@ import org.silkframework.rule.TransformSpec
 import org.silkframework.rule.execution.{ExecuteTransform, TransformReport}
 import org.silkframework.runtime.activity.{Activity, UserContext}
 import org.silkframework.runtime.plugin.PluginContext
-import org.silkframework.runtime.plugin.annotations.{Param, Plugin}
-import org.silkframework.runtime.plugin.types.IntOptionParameter
+import org.silkframework.runtime.plugin.annotations.Plugin
 import org.silkframework.workspace.ProjectTask
 import org.silkframework.workspace.activity.TaskActivityFactory
 import org.silkframework.workspace.activity.transform.TransformTaskUtils._
@@ -17,8 +16,7 @@ import org.silkframework.workspace.activity.transform.TransformTaskUtils._
   categories = Array("TransformSpecification"),
   description = "Executes the transformation."
 )
-case class ExecuteTransformFactory(@Param("Limits the maximum number of entities that are transformed.")
-                                   limit: IntOptionParameter = None) extends TaskActivityFactory[TransformSpec, ExecuteTransform] {
+case class ExecuteTransformFactory() extends TaskActivityFactory[TransformSpec, ExecuteTransform] {
 
   override def apply(task: ProjectTask[TransformSpec]): Activity[TransformReport] = {
     Activity.regenerating {
@@ -29,8 +27,7 @@ case class ExecuteTransformFactory(@Param("Limits the maximum number of entities
         (userContext: UserContext) => task.dataSource(userContext),
         (userContext: UserContext) => new CombinedEntitySink(task.entitySink(userContext).toSeq),
         (userContext: UserContext) => task.errorEntitySink(userContext),
-        (userContext: UserContext) => PluginContext.fromProject(task.project)(userContext),
-        limit
+        (userContext: UserContext) => PluginContext.fromProject(task.project)(userContext)
       )
     }
   }