Merge pull request #71 from vitrivr/feature/68

PR for Changes for issue 68

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/config/IndexContextFactory.kt

@@ -14,15 +14,16 @@ object IndexContextFactory {
     /**
      * Primarily creates the [ContentFactoriesFactory], on which foundation the [IndexContext] is created.
      *
-     * @param schema The [Schema] to this [IndexContext] to-be-built is for
      * @param contextConfig The [IngestionContextConfig] describing the [IndexContext] to-be-built.
      *
      * @return A [IndexContext] based on the [contextConfig]'s description and for the [schema].
      */
-    fun newContext(schema: Schema, contextConfig: IngestionContextConfig): IndexContext {
+    fun newContext(contextConfig: IngestionContextConfig): IndexContext {
         /* Load content factory. */
         val contentFactory = loadServiceForName<ContentFactoriesFactory>(contextConfig.contentFactory) ?: throw IllegalArgumentException("Failed to find content factory implementation for name '${contextConfig.contentFactory}'.")
 
+        val schema = contextConfig.schema
+
         /* Return new context. */
         return IndexContext(
             schema,

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/config/ingest/IngestionPipelineBuilder.kt

@@ -36,10 +36,10 @@ private val logger: KLogger = KotlinLogging.logger { }
  *  @author Ralph Gasser
  *  @version 2.0.0
  */
-class IngestionPipelineBuilder(val schema: Schema, val config: IngestionConfig) {
+class IngestionPipelineBuilder(val config: IngestionConfig) {
 
     /** The [IndexContext] */
-    private val context = IndexContextFactory.newContext(schema, config.context)
+    private val context = IndexContextFactory.newContext(config.context)
 
     /**
      * Build the indexing based this [IngestionPipelineBuilder]'s [config].

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/config/pipeline/execution/ExecutionServer.kt

@@ -121,10 +121,10 @@ class ExecutionServer {
      * @param query The [Operator] to execute.
      * @return The resulting [List] of [Retrieved]
      */
-    fun query(query: Operator<Retrieved>): List<Retrieved> {
+    fun query(query: Operator<out Retrievable>): List<Retrievable> {
         val jobId = UUID.randomUUID()
         val scope = CoroutineScope(this@ExecutionServer.dispatcher) + CoroutineName("query-job-$jobId")
-        val results = LinkedList<Retrieved>()
+        val results = LinkedList<Retrievable>()
         runBlocking {
             val job = scope.launch {
                 query.toFlow(scope).toList(results)

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/context/Context.kt

@@ -1,6 +1,8 @@
 package org.vitrivr.engine.core.context
 
 import kotlinx.serialization.Serializable
+import kotlinx.serialization.Transient
+import org.vitrivr.engine.core.model.metamodel.Schema
 
 /**
  * A [Context] provides additional contextual configuration in the form of key-value pairs,
@@ -51,4 +53,8 @@ sealed class Context() {
      * @return Either the value named [property] for the [operator] or [default]
      */
     fun getPropertyOrDefault(operator: String, property: String, default: String): String = getProperty(operator, property) ?: default
+
+
+    @Transient
+    abstract val schema: Schema
 }

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/context/IndexContext.kt

@@ -1,5 +1,6 @@
 package org.vitrivr.engine.core.context
 
+import kotlinx.serialization.Transient
 import org.vitrivr.engine.core.model.content.factory.ContentFactory
 import org.vitrivr.engine.core.model.metamodel.Schema
 import org.vitrivr.engine.core.resolver.Resolver
@@ -11,7 +12,8 @@ import org.vitrivr.engine.core.resolver.Resolver
  * @version 1.0.0
  */
 data class IndexContext(
-    val schema: Schema,
+    @Transient
+    override val schema: Schema,
 
     val contentFactory: ContentFactory,
 

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/context/IngestionContextConfig.kt

@@ -1,7 +1,9 @@
 package org.vitrivr.engine.core.context
 
 import kotlinx.serialization.Serializable
+import kotlinx.serialization.Transient
 import org.vitrivr.engine.core.model.content.factory.ContentFactory
+import org.vitrivr.engine.core.model.metamodel.Schema
 import org.vitrivr.engine.core.resolver.Resolver
 
 /**
@@ -12,6 +14,7 @@ import org.vitrivr.engine.core.resolver.Resolver
  */
 @Serializable
 class IngestionContextConfig(
+
     /** The simple or fully qualified class name of the [ContentFactory] to be used to construct the [IndexContext] */
     val contentFactory: String,
 
@@ -23,4 +26,9 @@ class IngestionContextConfig(
 
     /** [Map] of global parameters. */
     override val global: Map<String, String> = emptyMap()
-) : Context()
+) : Context() {
+
+    @Transient
+    override lateinit var schema: Schema
+
+}

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/context/QueryContext.kt

@@ -1,10 +1,15 @@
 package org.vitrivr.engine.core.context
 
 import kotlinx.serialization.Serializable
+import kotlinx.serialization.Transient
+import org.vitrivr.engine.core.model.metamodel.Schema
 
 @Serializable
 class QueryContext(
     override val local: Map<String, Map<String, String>> = emptyMap(),
     override val global: Map<String, String> = emptyMap()
 ) : Context() {
+
+    @Transient
+    override lateinit var schema: Schema
 }

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/model/metamodel/Schema.kt

@@ -89,7 +89,8 @@ open class Schema(val name: String = "vitrivr", val connection: Connection) : Cl
      * @param config The actual [IngestionConfig]
      */
     fun addIngestionPipeline(name: String, config: IngestionConfig) {
-        ingestionPipelineBuilders[name] = IngestionPipelineBuilder(this, config)
+        config.context.schema = this
+        ingestionPipelineBuilders[name] = IngestionPipelineBuilder(config)
     }
 
     /**

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/model/metamodel/SchemaManager.kt

@@ -69,12 +69,12 @@ class SchemaManager {
             )
         }
         config.extractionPipelines.map {
-            val indexConfig = IngestionConfig.read(Paths.get(it.path))
+            val ingestionConfig = IngestionConfig.read(Paths.get(it.path))
                 ?: throw IllegalArgumentException("Failed to read pipeline configuration from '${it.path}'.")
-            if (indexConfig.schema != schema.name) {
-                throw IllegalArgumentException("Schema name in pipeline configuration '${indexConfig.schema}' does not match schema name '${schema.name}'.")
+            if (ingestionConfig.schema != schema.name) {
+                throw IllegalArgumentException("Schema name in pipeline configuration '${ingestionConfig.schema}' does not match schema name '${schema.name}'.")
             }
-            schema.addIngestionPipeline(it.name, indexConfig)
+            schema.addIngestionPipeline(it.name, ingestionConfig)
         }
 
         /* Cache and return connection. */

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/model/retrievable/Ingested.kt

@@ -7,4 +7,7 @@ package org.vitrivr.engine.core.model.retrievable
  * @author Luca Rossetto
  * @version 1.2.0
  */
-class Ingested(override val id: RetrievableId, override val type: String?, override val transient: Boolean) : AbstractRetrievable(id, type, transient)
+data class Ingested(override val id: RetrievableId, override val type: String?, override val transient: Boolean) : AbstractRetrievable(id, type, transient) {
+
+    override fun copy(): Ingested = this.copy(id = id)
+}

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/model/retrievable/Retrievable.kt

@@ -151,4 +151,9 @@ interface Retrievable : Persistable {
      * @return List of matching [Relationship]s
      */
     fun findRelationship(predicate: Predicate<Relationship>): List<Relationship> = this.relationships.filter { predicate.test(it) }
+
+    /**
+     * Creates a copy of this [Retrievable]
+     */
+    fun copy(): Retrievable
 }

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/model/retrievable/Retrieved.kt

@@ -13,4 +13,6 @@ data class Retrieved(override val id: UUID, override val type: String?, override
     constructor(retrievable: Retrievable) : this(retrievable.id, retrievable.type, retrievable.transient) {
         retrievable.attributes.forEach { this.addAttribute(it) }
     }
+
+    override fun copy(): Retrieved = this.copy(id = id)
 }

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/operators/Operator.kt

@@ -38,7 +38,7 @@ sealed interface Operator<O> {
      */
     interface Unary<I,O>: Operator<O> {
         /** The input [Operator]. */
-        val input: Operator<I>
+        val input: Operator<out I>
 
         /**
          * Returns root of this [Unary].
@@ -51,10 +51,10 @@ sealed interface Operator<O> {
      */
     interface Binary<I,O>: Operator<O> {
         /** The input [Operator]. */
-        val input1: Operator<I>
+        val input1: Operator<out I>
 
         /** The input [Operator]. */
-        val input2: Operator<I>
+        val input2: Operator<out I>
 
         /**
          * Returns root of this [Unary], which is the left-hand operator (by definition).
@@ -67,7 +67,7 @@ sealed interface Operator<O> {
      */
     interface NAry<I,O>: Operator<O> {
         /** The input [Operator]s. */
-        val inputs: List<Operator<I>>
+        val inputs: List<Operator<out I>>
 
         /**
          * Returns root of this [Unary], which is the left-hand operator (by definition).
@@ -80,7 +80,7 @@ sealed interface Operator<O> {
      */
     interface Sink<I>: Operator<Unit> {
         /** The input [Operator]. */
-        val input: Operator<I>
+        val input: Operator<out I>
 
         /**
          * Returns root of this [Unary], which is the left-hand operator (by definition).

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/operators/general/Aggregator.kt

@@ -0,0 +1,6 @@
+package org.vitrivr.engine.core.operators.general
+
+import org.vitrivr.engine.core.model.retrievable.Retrievable
+import org.vitrivr.engine.core.operators.Operator
+
+interface Aggregator : Operator.NAry<Retrievable, Retrievable>

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/operators/general/AggregatorFactory.kt

@@ -0,0 +1,11 @@
+package org.vitrivr.engine.core.operators.general
+
+import org.vitrivr.engine.core.context.Context
+import org.vitrivr.engine.core.model.retrievable.Retrievable
+import org.vitrivr.engine.core.operators.Operator
+
+interface AggregatorFactory {
+
+    fun newAggregator(name: String, inputs: List<Operator<out Retrievable>>, context: Context): Aggregator
+
+}

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/operators/general/TransformerFactory.kt

@@ -1,6 +1,6 @@
 package org.vitrivr.engine.core.operators.general
 
-import org.vitrivr.engine.core.context.IndexContext
+import org.vitrivr.engine.core.context.Context
 import org.vitrivr.engine.core.model.retrievable.Retrievable
 import org.vitrivr.engine.core.operators.Operator
 
@@ -16,7 +16,7 @@ interface TransformerFactory {
      *
      * @param name The name of the [Transformer]
      * @param input The input [Operator].
-     * @param context The [IndexContext] to use.
+     * @param context The [Context] to use.
      */
-    fun newTransformer(name: String, input: Operator<Retrievable>, context: IndexContext): Transformer
+    fun newTransformer(name: String, input: Operator<out Retrievable>, context: Context): Transformer
 }

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/operators/transform/filter/AbstractFilterTransformer.kt

@@ -13,7 +13,7 @@ import org.vitrivr.engine.core.operators.general.Transformer
  * @author Ralph Gasser
  * @version 1.0.0
  */
-abstract class AbstractFilterTransformer(override val input: Operator<Retrievable>, inline val predicate: (Retrievable) -> Boolean) : Transformer {
+abstract class AbstractFilterTransformer(override val input: Operator<out Retrievable>, inline val predicate: (Retrievable) -> Boolean) : Transformer {
     override fun toFlow(scope: CoroutineScope): Flow<Retrievable> = this.input.toFlow(scope).filter {
         this.predicate(it)
     }

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/operators/transform/filter/DistinctTransformer.kt

@@ -3,7 +3,7 @@ package org.vitrivr.engine.core.operators.transform.filter
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.channelFlow
-import org.vitrivr.engine.core.context.IndexContext
+import org.vitrivr.engine.core.context.Context
 import org.vitrivr.engine.core.model.relationship.Relationship
 import org.vitrivr.engine.core.model.retrievable.Retrievable
 import org.vitrivr.engine.core.operators.Operator
@@ -20,12 +20,12 @@ import java.util.*
  * @version 1.0.0
  */
 class DistinctTransformer : TransformerFactory {
-    override fun newTransformer(name: String, input: Operator<Retrievable>, context: IndexContext): Transformer = Instance(input)
+    override fun newTransformer(name: String, input: Operator<out Retrievable>, context: Context): Transformer = Instance(input)
 
     /**
      * [Transformer] that extracts [Retrievable] objects from a [Flow] of [Retrievable] objects based on a given [Relationship].
      */
-    private class Instance(override val input: Operator<Retrievable>) : Transformer {
+    private class Instance(override val input: Operator<out Retrievable>) : Transformer {
         override fun toFlow(scope: CoroutineScope): Flow<Retrievable> = channelFlow {
             val set = HashSet<UUID>()
             this@Instance.input.toFlow(scope).collect {

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/operators/transform/filter/TypeFilterTransformer.kt

@@ -1,6 +1,6 @@
 package org.vitrivr.engine.core.operators.transform.filter
 
-import org.vitrivr.engine.core.context.IndexContext
+import org.vitrivr.engine.core.context.Context
 import org.vitrivr.engine.core.model.retrievable.Ingested
 import org.vitrivr.engine.core.model.retrievable.Retrievable
 import org.vitrivr.engine.core.operators.Operator
@@ -14,10 +14,10 @@ import org.vitrivr.engine.core.operators.general.TransformerFactory
  * @version 1.0.0
  */
 class TypeFilterTransformer : TransformerFactory {
-    override fun newTransformer(name: String, input: Operator<Retrievable>, context: IndexContext): Transformer {
+    override fun newTransformer(name: String, input: Operator<out Retrievable>, context: Context): Transformer {
         val predicate = context[name, "type"] ?: throw IllegalArgumentException("The type filter transformer requires a type name.")
         return Instance(input, predicate)
     }
 
-    private class Instance(input: Operator<Retrievable>, val type: String) : AbstractFilterTransformer(input, { it.type == type })
+    private class Instance(input: Operator<out Retrievable>, val type: String) : AbstractFilterTransformer(input, { it.type == type })
 }

vitrivr-engine-core/src/main/kotlin/org/vitrivr/engine/core/operators/transform/map/MapRelationshipTransformer.kt

@@ -3,7 +3,7 @@ package org.vitrivr.engine.core.operators.transform.map
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.channelFlow
-import org.vitrivr.engine.core.context.IndexContext
+import org.vitrivr.engine.core.context.Context
 import org.vitrivr.engine.core.model.relationship.Relationship
 import org.vitrivr.engine.core.model.retrievable.Ingested
 import org.vitrivr.engine.core.model.retrievable.Retrievable
@@ -18,15 +18,15 @@ import org.vitrivr.engine.core.operators.general.TransformerFactory
  * @version 1.0.0
  */
 class MapRelationshipTransformer : TransformerFactory {
-    override fun newTransformer(name: String, input: Operator<Retrievable>, context: IndexContext): Transformer {
+    override fun newTransformer(name: String, input: Operator<out Retrievable>, context: Context): Transformer {
         val predicate = context[name, "predicate"] ?: throw IllegalArgumentException("The relationship transformer requires a predicate to be specified.")
         return Instance(input, predicate)
     }
 
     /**
      * [Transformer] that extracts [Ingested] objects from a [Flow] of [Ingested] objects based on a given [Relationship].
      */
-    private class Instance(override val input: Operator<Retrievable>, val name: String) : Transformer {
+    private class Instance(override val input: Operator<out Retrievable>, val name: String) : Transformer {
         override fun toFlow(scope: CoroutineScope): Flow<Ingested> = channelFlow {
             this@Instance.input.toFlow(scope).collect { ingested ->
                 ingested.relationships.filter { it.predicate == this@Instance.name }.forEach { relationship ->