Filter by type in the SQL query for Composite Views (#4231)

BlueBrain · Aug 31, 2023 · f7b24c6 · f7b24c6
1 parent 89edef4
commit f7b24c6
Show file tree

Hide file tree

Showing 6 changed files with 59 additions and 37 deletions.
diff --git a/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/routes/ElemRoutes.scala b/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/routes/ElemRoutes.scala
@@ -18,7 +18,7 @@ import ch.epfl.bluebrain.nexus.delta.sdk.model.BaseUri
 import ch.epfl.bluebrain.nexus.delta.sdk.permissions.Permissions.events
 import ch.epfl.bluebrain.nexus.delta.sdk.sse.SseElemStream
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.ProjectRef
-import ch.epfl.bluebrain.nexus.delta.sourcing.model.Tag.UserTag
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.Tag.{Latest, UserTag}
 import ch.epfl.bluebrain.nexus.delta.sourcing.query.SelectFilter
 import ch.epfl.bluebrain.nexus.delta.sourcing.stream.RemainingElems
 import io.circe.syntax.EncoderOps
@@ -54,30 +54,32 @@ class ElemRoutes(
           pathPrefix("elems") {
             resolveProjectRef { project =>
               authorizeFor(project, events.read).apply {
-                concat(
-                  (get & pathPrefix("continuous") & parameter("tag".as[UserTag].?)) { tag =>
-                    operationName(s"$prefixSegment/$project/elems/continuous") {
-                      emit(sseElemStream.continuous(project, SelectFilter.tagOrLatest(tag), offset))
+                (parameter("tag".as[UserTag].?) & types(project)) { (tag, types) =>
+                  concat(
+                    (get & pathPrefix("continuous")) {
+                      operationName(s"$prefixSegment/$project/elems/continuous") {
+                        emit(sseElemStream.continuous(project, SelectFilter(types, tag.getOrElse(Latest)), offset))
+                      }
+                    },
+                    (get & pathPrefix("currents")) {
+                      operationName(s"$prefixSegment/$project/elems/currents") {
+                        emit(sseElemStream.currents(project, SelectFilter(types, tag.getOrElse(Latest)), offset))
+                      }
+                    },
+                    (get & pathPrefix("remaining")) {
+                      operationName(s"$prefixSegment/$project/elems/remaining") {
+                        emit(
+                          sseElemStream.remaining(project, SelectFilter(types, tag.getOrElse(Latest)), offset).map {
+                            r => r.getOrElse(RemainingElems(0L, Instant.EPOCH))
+                          }
+                        )
+                      }
+                    },
+                    head {
+                      complete(OK)
                     }
-                  },
-                  (get & pathPrefix("currents") & parameter("tag".as[UserTag].?)) { tag =>
-                    operationName(s"$prefixSegment/$project/elems/currents") {
-                      emit(sseElemStream.currents(project, SelectFilter.tagOrLatest(tag), offset))
-                    }
-                  },
-                  (get & pathPrefix("remaining") & parameter("tag".as[UserTag].?)) { tag =>
-                    operationName(s"$prefixSegment/$project/elems/remaining") {
-                      emit(
-                        sseElemStream.remaining(project, SelectFilter.tagOrLatest(tag), offset).map { r =>
-                          r.getOrElse(RemainingElems(0L, Instant.EPOCH))
-                        }
-                      )
-                    }
-                  },
-                  head {
-                    complete(OK)
-                  }
-                )
+                  )
+                }
               }
             }
           }

diff --git a/.../main/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/client/DeltaClient.scala b/.../main/scala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/client/DeltaClient.scala
@@ -145,8 +145,14 @@ object DeltaClient {
         }
     }
 
+    private def typeQuery(types: Set[Iri])               =
+      if (types.isEmpty) Query.Empty
+      else Query(types.map(t => "type" -> t.toString).toList: _*)
+
     private def elemAddress(source: RemoteProjectSource) =
-      source.endpoint / "elems" / source.project.organization.value / source.project.project.value
+      (source.endpoint / "elems" / source.project.organization.value / source.project.project.value)
+        .withQuery(Query("tag" -> source.selectFilter.tag.toString))
+        .withQuery(typeQuery(source.selectFilter.types))
 
     override def resourceAsNQuads(source: RemoteProjectSource, id: Iri): HttpResult[Option[NQuads]] = {
       implicit val cred: Option[AuthToken] = token(source)

diff --git a/...cala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/model/CompositeViewSource.scala b/...cala/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/model/CompositeViewSource.scala
@@ -11,7 +11,8 @@ import ch.epfl.bluebrain.nexus.delta.rdf.jsonld.decoder.semiauto.deriveDefaultJs
 import ch.epfl.bluebrain.nexus.delta.sdk.model.BaseUri
 import ch.epfl.bluebrain.nexus.delta.sdk.instances._
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.{Identity, ProjectRef}
-import ch.epfl.bluebrain.nexus.delta.sourcing.model.Tag.UserTag
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.Tag.{Latest, UserTag}
+import ch.epfl.bluebrain.nexus.delta.sourcing.query.SelectFilter
 import ch.epfl.bluebrain.nexus.delta.sourcing.stream.PipeChain
 import io.circe.{Encoder, Json}
 
@@ -54,6 +55,13 @@ sealed trait CompositeViewSource extends Product with Serializable {
     */
   def resourceTag: Option[UserTag]
 
+  /**
+    * @return
+    *   the [[SelectFilter]] for the given view; used to filter the data that is indexed
+    */
+  def selectFilter: SelectFilter =
+    SelectFilter(resourceTypes, resourceTag.getOrElse(Latest))
+
   /**
     * @return
     *   whether to consider deprecated resources for indexing

diff --git a/...la/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/stream/CompositeGraphStream.scala b/...la/ch/epfl/bluebrain/nexus/delta/plugins/compositeviews/stream/CompositeGraphStream.scala
@@ -5,7 +5,6 @@ import ch.epfl.bluebrain.nexus.delta.plugins.compositeviews.model.CompositeViewS
 import ch.epfl.bluebrain.nexus.delta.sdk.stream.GraphResourceStream
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.{ElemPipe, ProjectRef}
 import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset
-import ch.epfl.bluebrain.nexus.delta.sourcing.query.SelectFilter
 import ch.epfl.bluebrain.nexus.delta.sourcing.state.GraphResource
 import ch.epfl.bluebrain.nexus.delta.sourcing.stream.{RemainingElems, Source}
 import io.circe.Json
@@ -56,27 +55,27 @@ object CompositeGraphStream {
     override def main(source: CompositeViewSource, project: ProjectRef): Source = {
       source match {
         case p: ProjectSource       =>
-          Source(local.continuous(project, SelectFilter.tagOrLatest(p.resourceTag), _).through(drainSource))
+          Source(local.continuous(project, p.selectFilter, _).through(drainSource))
         case c: CrossProjectSource  =>
-          Source(local.continuous(c.project, SelectFilter.tagOrLatest(c.resourceTag), _).through(drainSource))
+          Source(local.continuous(c.project, c.selectFilter, _).through(drainSource))
         case r: RemoteProjectSource => remote.main(r)
       }
     }
 
     override def rebuild(source: CompositeViewSource, project: ProjectRef): Source = {
       source match {
         case p: ProjectSource       =>
-          Source(local.currents(project, SelectFilter.tagOrLatest(p.resourceTag), _).through(drainSource))
+          Source(local.currents(project, p.selectFilter, _).through(drainSource))
         case c: CrossProjectSource  =>
-          Source(local.currents(c.project, SelectFilter.tagOrLatest(c.resourceTag), _).through(drainSource))
+          Source(local.currents(c.project, c.selectFilter, _).through(drainSource))
         case r: RemoteProjectSource => remote.rebuild(r)
       }
     }
 
     override def remaining(source: CompositeViewSource, project: ProjectRef): Offset => UIO[Option[RemainingElems]] =
       source match {
-        case p: ProjectSource       => local.remaining(project, SelectFilter.tagOrLatest(p.resourceTag), _)
-        case c: CrossProjectSource  => local.remaining(c.project, SelectFilter.tagOrLatest(c.resourceTag), _)
+        case p: ProjectSource       => local.remaining(project, p.selectFilter, _)
+        case c: CrossProjectSource  => local.remaining(c.project, c.selectFilter, _)
         case r: RemoteProjectSource => remote.remaining(r, _).map(Some(_))
       }
   }

diff --git a/...rcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/query/SelectFilter.scala b/...rcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/query/SelectFilter.scala
@@ -20,8 +20,4 @@ object SelectFilter {
   /** All types with latest tag */
   val latest: SelectFilter = SelectFilter(Set.empty, Tag.Latest)
 
-  /** All types with specified tag if it exists, otherwise latest */
-  val tagOrLatest: Option[Tag] => SelectFilter =
-    tag => SelectFilter(Set.empty, tag.getOrElse(Tag.Latest))
-
 }
diff --git a/docs/src/main/paradox/docs/releases/v1.9-release-notes.md b/docs/src/main/paradox/docs/releases/v1.9-release-notes.md
@@ -44,10 +44,21 @@ It is now possible to aggregate resources by `@type` or `project`.
 
 ### Views
 
+#### Indexing errors listing
+
 Indexing errors can now be listed and filtered for a given view.
 
 @ref:[More information](../delta/api/views/index.md#listing-indexing-failures)
 
+#### Resource type filtering performance improvement 
+
+To improve indexing performance, the types defined in the 
+@ref:[FilterByType pipe](../delta/api/views/pipes.md#filter-by-type), 
+@ref:[Sparql View payload](../delta/api/views/sparql-view-api.md#payload), or the 
+@ref:[Composite View source payload](../delta/api/views/composite-view-api.md#sources) 
+are filtered in PostgreSQL rather than in Nexus Delta.
+This avoids querying for data just to discard it straight away.
+
 ### Composite views
 
 To enhance performance of indexing of composite views, Nexus Delta introduces the following features.