-
Notifications
You must be signed in to change notification settings - Fork 74
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Apply indexing rev rules to composite view indexing (#4168)
* Apply indexing rev rules to composite view indexing --------- Co-authored-by: Simon Dumas <simon.dumas@epfl.ch>
- Loading branch information
Showing
24 changed files
with
874 additions
and
224 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
63 changes: 63 additions & 0 deletions
63
.../scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/indexing/CompositeSinks.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.indexing | ||
|
|
||
| import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.client.BlazegraphClient | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.indexing.BlazegraphSink | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.CompositeSink | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.config.CompositeViewsConfig | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.indexing.CompositeViewDef.ActiveViewDef | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.model.CompositeViewProjection | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.model.CompositeViewProjection.{ElasticSearchProjection, SparqlProjection} | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.ElasticSearchClient | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.jsonld.context.RemoteContextResolution | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.model.BaseUri | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.stream.Operation.Sink | ||
|
|
||
| /** | ||
| * Defines the sinks for the indexing progress for a composite view | ||
| */ | ||
| trait CompositeSinks { | ||
|
|
||
| /** | ||
| * The sink for the current namespace | ||
| */ | ||
| def commonSink(view: ActiveViewDef): Sink | ||
|
|
||
| /** | ||
| * The sink for a given projection | ||
| */ | ||
| def projectionSink(view: ActiveViewDef, target: CompositeViewProjection): Sink | ||
| } | ||
|
|
||
| object CompositeSinks { | ||
|
|
||
| def apply( | ||
| prefix: String, | ||
| esClient: ElasticSearchClient, | ||
| blazeClient: BlazegraphClient, | ||
| cfg: CompositeViewsConfig | ||
| )(implicit base: BaseUri, rcr: RemoteContextResolution): CompositeSinks = new CompositeSinks { | ||
|
|
||
| /** | ||
| * The sink for the current namespace | ||
| */ | ||
| override def commonSink(view: ActiveViewDef): Sink = { | ||
| val common = commonNamespace(view.uuid, view.indexingRev, prefix) | ||
| BlazegraphSink(blazeClient, cfg.blazegraphBatch, common) | ||
| } | ||
|
|
||
| /** | ||
| * The sink for a given projection | ||
| */ | ||
| override def projectionSink(view: ActiveViewDef, target: CompositeViewProjection): Sink = { | ||
| val common = commonNamespace(view.uuid, view.indexingRev, prefix) | ||
| target match { | ||
| case e: ElasticSearchProjection => | ||
| val index = projectionIndex(e, view.uuid, prefix) | ||
| CompositeSink.elasticSink(blazeClient, esClient, index, common, cfg).apply(e) | ||
| case s: SparqlProjection => | ||
| val namespace = projectionNamespace(s, view.uuid, prefix) | ||
| CompositeSink.blazeSink(blazeClient, namespace, common, cfg).apply(s) | ||
| } | ||
| } | ||
| } | ||
| } |
129 changes: 54 additions & 75 deletions
129
...scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/indexing/CompositeSpaces.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,98 +1,77 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.indexing | ||
|
|
||
| import cats.syntax.all._ | ||
| import ch.epfl.bluebrain.nexus.delta.kernel.Logger | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.client.BlazegraphClient | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.indexing.BlazegraphSink | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.CompositeSink | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.config.CompositeViewsConfig | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.indexing.CompositeViewDef.ActiveViewDef | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.model.CompositeViewProjection | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.model.CompositeViewProjection.{ElasticSearchProjection, SparqlProjection} | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.{ElasticSearchClient, IndexLabel} | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.IriOrBNode.Iri | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.jsonld.context.RemoteContextResolution | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.model.BaseUri | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.stream.Operation.Sink | ||
| import com.typesafe.scalalogging.Logger | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.ElasticSearchClient | ||
| import monix.bio.Task | ||
|
|
||
| /** | ||
| * Defines the pipes, sinks for the indexing progress as well as the init and destroy tasks for a composite view | ||
| * @param init | ||
| * the task to create the different namespaces and indices | ||
| * @param destroy | ||
| * the task to destroy the different namespaces and indices | ||
| * @param commonSink | ||
| * the sink for the common sparql space | ||
| * @param targetSink | ||
| * the function to create a sink for a [[CompositeViewProjection]] | ||
| * Defines the operations to create and destroy the namespaces of a composite view | ||
| */ | ||
| final case class CompositeSpaces( | ||
| init: Task[Unit], | ||
| destroy: Task[Unit], | ||
| commonSink: Sink, | ||
| targetSink: CompositeViewProjection => Sink | ||
| ) | ||
| trait CompositeSpaces { | ||
|
|
||
| object CompositeSpaces { | ||
|
|
||
| private val logger: Logger = Logger[CompositeSpaces] | ||
| /** | ||
| * Creates all spaces for the given view | ||
| */ | ||
| def init(view: ActiveViewDef): Task[Unit] | ||
|
|
||
| trait Builder { | ||
| /** | ||
| * Destroys all spaces for the given view | ||
| */ | ||
| def destroyAll(view: ActiveViewDef): Task[Unit] | ||
|
|
||
| /** | ||
| * Compute the spaces for the given view | ||
| * @param view | ||
| * the active view | ||
| */ | ||
| def apply(view: ActiveViewDef): CompositeSpaces | ||
| } | ||
| /** | ||
| * Destroys space for the projection of the given view | ||
| */ | ||
| def destroyProjection(view: ActiveViewDef, projection: CompositeViewProjection): Task[Unit] | ||
|
|
||
| object Builder { | ||
| def apply( | ||
| prefix: String, | ||
| esClient: ElasticSearchClient, | ||
| blazeClient: BlazegraphClient, | ||
| cfg: CompositeViewsConfig | ||
| )(implicit base: BaseUri, rcr: RemoteContextResolution): CompositeSpaces.Builder = (view: ActiveViewDef) => { | ||
| } | ||
|
|
||
| // Operations and sinks for the common space | ||
| val common = commonNamespace(view.uuid, view.indexingRev, prefix) | ||
| val commonSink = BlazegraphSink(blazeClient, cfg.blazegraphBatch, common) | ||
| val createCommon = blazeClient.createNamespace(common) | ||
| val deleteCommon = blazeClient.deleteNamespace(common) | ||
| object CompositeSpaces { | ||
|
|
||
| // Create sinks | ||
| def createBlazeSink(namespace: String): SparqlProjection => Sink = | ||
| CompositeSink.blazeSink(blazeClient, namespace, common, cfg) | ||
| def createEsSink(index: IndexLabel): ElasticSearchProjection => Sink = | ||
| CompositeSink.elasticSink(blazeClient, esClient, index, common, cfg) | ||
| private val logger: Logger = Logger[CompositeSpaces] | ||
|
|
||
| // Compute the init and destroy operations as well as the sink for the different projections of the composite views | ||
| val start: (Task[Unit], Task[Unit], Map[Iri, Sink]) = (createCommon.void, deleteCommon.void, Map.empty[Iri, Sink]) | ||
| val (init, destroy, sinkMap) = view.value.projections.foldLeft(start) { | ||
| case ((create, delete, sinkMap), p: ElasticSearchProjection) => | ||
| val index = projectionIndex(p, view.uuid, prefix) | ||
| ( | ||
| create >> esClient.createIndex(index, Some(p.mapping), p.settings).void, | ||
| delete >> esClient.deleteIndex(index).void, | ||
| sinkMap.updated(p.id, createEsSink(index)(p)) | ||
| ) | ||
| case ((create, delete, sinkMap), s: SparqlProjection) => | ||
| def apply( | ||
| prefix: String, | ||
| esClient: ElasticSearchClient, | ||
| blazeClient: BlazegraphClient | ||
| ): CompositeSpaces = new CompositeSpaces { | ||
| override def init(view: ActiveViewDef): Task[Unit] = { | ||
| val common = commonNamespace(view.uuid, view.indexingRev, prefix) | ||
| val createCommon = blazeClient.createNamespace(common).void | ||
| val result = view.value.projections.foldLeft[Task[Unit]](createCommon) { | ||
| case (acc, e: ElasticSearchProjection) => | ||
| val index = projectionIndex(e, view.uuid, prefix) | ||
| acc >> esClient.createIndex(index, Some(e.mapping), e.settings).void | ||
| case (acc, s: SparqlProjection) => | ||
| val namespace = projectionNamespace(s, view.uuid, prefix) | ||
| ( | ||
| create >> blazeClient.createNamespace(namespace).void, | ||
| delete >> blazeClient.deleteNamespace(namespace).void, | ||
| sinkMap.updated(s.id, createBlazeSink(namespace)(s)) | ||
| ) | ||
| acc >> blazeClient.createNamespace(namespace).void | ||
| } | ||
| logger.debug(s"Creating namespaces and indices for composite view ${view.ref}") >> result | ||
| } | ||
|
|
||
| CompositeSpaces( | ||
| Task.delay(logger.debug("Creating namespaces and indices for composite view {}", view.ref)) >> init, | ||
| Task.delay(logger.debug("Deleting namespaces and indices for composite view {}", view.ref)) >> destroy, | ||
| commonSink, | ||
| (p: CompositeViewProjection) => sinkMap(p.id) | ||
| ) | ||
| override def destroyAll(view: ActiveViewDef): Task[Unit] = { | ||
| val common = commonNamespace(view.uuid, view.indexingRev, prefix) | ||
| val deleteCommon = blazeClient.deleteNamespace(common).void | ||
| val result = view.value.projections.foldLeft[Task[Unit]](deleteCommon) { case (acc, p) => | ||
| acc >> destroyProjection(view, p) | ||
| } | ||
| logger.debug(s"Deleting namespaces and indices for composite view ${view.ref}") >> result | ||
| } | ||
|
|
||
| override def destroyProjection(view: ActiveViewDef, projection: CompositeViewProjection): Task[Unit] = | ||
| logger.debug(s"Deleting namespace/index for projection ${projection.id} of composite view ${view.ref}") >> { | ||
| projection match { | ||
| case e: ElasticSearchProjection => | ||
| val index = projectionIndex(e, view.uuid, prefix) | ||
| esClient.deleteIndex(index).void | ||
| case s: SparqlProjection => | ||
| val namespace = projectionNamespace(s, view.uuid, prefix) | ||
| blazeClient.deleteNamespace(namespace).void | ||
| } | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.