Repository: holunda-io/camunda-bpm-taskpool (kotlin)

Issue:
Filters on payload attributes cause multiple join of the same table
### Steps to reproduce

* camunda-bpm-taskpool version: any
* Camunda BPM version: any
* JDK version:  any
* Operating system: any
* Steps: (what exactly are you doing with the above reproducer?)
  * Create a TaskQuery for a JPA View that has multiple filters for payload attributes


### Expected behaviour
In order to filter based on payload attributes one join between the `plf_task` and `plf_task_payload_attributes` tables has to be performed. On this join result all the filter criteria for payload attributes can be evaluated.

### Actual behaviour
Each filter criterion is transformed into a specification by itself
```
internal fun Criterion.PayloadEntryCriterion.toTaskSpecification(): Specification<TaskEntity> {
  return when (this.operator) {
    EQUALS -> hasTaskPayloadAttribute(this.name, this.value)
    else -> throw IllegalArgumentException("JPA View currently supports only equals as operator for filtering of payload attributes.")
  }
}
```

and for each of these criteria a join is performed
```
    fun hasTaskPayloadAttribute(name: String, value: String): Specification<TaskEntity> =
      Specification { task, query, builder ->
        query.distinct(true)
        val join = task.join<TaskEntity, Set<PayloadAttribute>>(TaskEntity::payloadAttributes.name)
        val pathEquals = builder.equal(
          join.get<String>(PayloadAttribute::path.name),
          name
        )
        val valueEquals = builder.equal(
          join.get<String>(PayloadAttribute::value.name),
          value
        )
        builder.and(pathEquals, valueEquals)
      }
```

This results in the following pattern in the generated SQL
```
 select
        taskentity0_.task_id as task_id1_10_,
        taskentity0_.assignee_id as assignee2_10_,
        taskentity0_.business_key as business3_10_,
        taskentity0_.date_created as date_cre4_10_,
        taskentity0_.description as descript5_10_,
        taskentity0_.date_due as date_due6_10_,
        taskentity0_.date_follow_up as date_fol7_10_,
        taskentity0_.form_key as form_key8_10_,
        taskentity0_.name as name9_10_,
        taskentity0_.owner_id as owner_i10_10_,
        taskentity0_.payload as payload11_10_,
        taskentity0_.priority as priorit12_10_,
        taskentity0_.application_name as applica13_10_,
        taskentity0_.source_def_id as source_14_10_,
        taskentity0_.source_def_key as source_15_10_,
        taskentity0_.source_execution_id as source_16_10_,
        taskentity0_.source_instance_id as source_17_10_,
        taskentity0_.source_name as source_18_10_,
        taskentity0_.source_type as source_19_10_,
        taskentity0_.source_tenant_id as source_20_10_,
        taskentity0_.task_def_key as task_de21_10_ 
    from
        plf_task taskentity0_ 
    inner join
        plf_task_payload_attributes payloadatt1_ 
            on taskentity0_.task_id=payloadatt1_.task_id 
    inner join
        plf_task_payload_attributes payloadatt2_ 
            on taskentity0_.task_id=payloadatt2_.task_id 
    where
        (
            payloadatt1_.path=? 
            and payloadatt1_.value=? 
            or payloadatt2_.path=? 
            and payloadatt2_.value=?
        ) 
        and (
            ? in (
                select
                    authorized3_.authorized_principal 
                from
                    plf_task_authorizations authorized3_ 
                where
                    taskentity0_.task_id=authorized3_.task_id
            ) 
            or ? in (
                select
                    authorized4_.authorized_principal 
                from
                    plf_task_authorizations authorized4_ 
                where
                    taskentity0_.task_id=authorized4_.task_id
            ) 
            or taskentity0_.assignee_id=?
        ) 
    order by
        taskentity0_.date_created desc limit ?
```

In the above example two filter criteria on the task payload were given, which results in consecutive joins of the task and payload table.
`(plf_task join plf_task_payload_attributes) join plf_task_payload_attributes)`

This means that the resulting joined table grows exponentially with the number of filters, even though only one single join of these tables would be necessary.

One way to eliminate this would be to build the specification from the whole list of criteria on the payload and only joining once there. The question is how the signature for that method would look like. Because we wouldn't really want to determine the composition logic in the method so we would have to find a way to represent the composition in the parameters to make the conversion to the JPA specification more generic.

Pull request:
For filters on the same attribute only one join with the payload attribute table is performed now, instead of one join per possible value. fixes #849

Patch:
diff --git a/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/SpecificationExt.kt b/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/SpecificationExt.kt
index 4ce9f227f..81a3b9081 100644
--- a/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/SpecificationExt.kt
+++ b/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/SpecificationExt.kt
@@ -152,7 +152,7 @@ internal fun List<Criterion>.toDataEntryPayloadSpecification(): Specification<Da
   val relevant = this.filterIsInstance<Criterion.PayloadEntryCriterion>()
   // compose criteria with same name with OR and criteria with different names with AND
   val relevantByName = relevant.groupBy { it.name }
-  val orComposedByName = relevantByName.map { (_, criteria) -> composeOr(criteria.map { it.toDataEntrySpecification() }) }
+  val orComposedByName = relevantByName.map { (_, criteria) -> criteria.toOrDataEntrySpecification() }
 
   return composeAnd(orComposedByName)
 }
@@ -164,7 +164,7 @@ internal fun List<Criterion>.toTaskPayloadSpecification(): Specification<TaskEnt
   val relevant = this.filterIsInstance<Criterion.PayloadEntryCriterion>()
   // compose criteria with same name with OR and criteria with different names with AND
   val relevantByName = relevant.groupBy { it.name }
-  val orComposedByName = relevantByName.map { (_, criteria) -> composeOr(criteria.map { it.toTaskSpecification() }) }
+  val orComposedByName = relevantByName.map { (_, criteria) -> criteria.toOrTaskSpecification() }
 
   return composeAnd(orComposedByName)
 }
@@ -218,13 +218,15 @@ internal fun Criterion.TaskCriterion.toTaskSpecification(): Specification<TaskEn
 }
 
 /**
- * Creates JPA Specification for query of payload attributes based on JSON paths.
+ * Creates JPA Specification for query of payload attributes based on JSON paths. All criteria must have the same path
+ * and will be composed by the logical OR operator.
  */
-internal fun Criterion.PayloadEntryCriterion.toTaskSpecification(): Specification<TaskEntity> {
-  return when (this.operator) {
-    EQUALS -> hasTaskPayloadAttribute(this.name, this.value)
-    else -> throw IllegalArgumentException("JPA View currently supports only equals as operator for filtering of payload attributes.")
-  }
+internal fun List<Criterion.PayloadEntryCriterion>.toOrTaskSpecification(): Specification<TaskEntity> {
+  require(this.isNotEmpty()) { "List of criteria must not be empty." }
+  require(this.all { it.operator == EQUALS }) { "JPA View currently supports only equals as operator for filtering of payload attributes." }
+  require(this.distinctBy { it.name }.size == 1) { "All criteria must have the same path." }
+
+  return hasTaskPayloadAttribute(this.first().name, this.map { it.value })
 }
 
 /**
@@ -242,13 +244,15 @@ internal fun Criterion.DataEntryCriterion.toDataEntrySpecification(): Specificat
 }
 
 /**
- * Creates JPA Specification for query of payload attributes based on JSON paths.
+ * Creates JPA Specification for query of payload attributes based on JSON paths. All criteria must have the same path
+ * and will be composed by the logical OR operator.
  */
-internal fun Criterion.PayloadEntryCriterion.toDataEntrySpecification(): Specification<DataEntryEntity> {
-  return when (this.operator) {
-    EQUALS -> hasDataEntryPayloadAttribute(this.name, this.value)
-    else -> throw IllegalArgumentException("JPA View currently supports only equals as operator for filtering of payload attributes.")
-  }
+internal fun List<Criterion.PayloadEntryCriterion>.toOrDataEntrySpecification(): Specification<DataEntryEntity> {
+  require(this.isNotEmpty()) { "List of criteria must not be empty." }
+  require(this.all { it.operator == EQUALS }) { "JPA View currently supports only equals as operator for filtering of payload attributes." }
+  require(this.distinctBy { it.name }.size == 1) { "All criteria must have the same path." }
+
+  return hasDataEntryPayloadAttribute(this.first().name, this.map { it.value })
 }
 
 
diff --git a/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/data/DataEntryRepository.kt b/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/data/DataEntryRepository.kt
index 68f82ce7f..2d7354686 100644
--- a/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/data/DataEntryRepository.kt
+++ b/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/data/DataEntryRepository.kt
@@ -84,9 +84,10 @@ interface DataEntryRepository : CrudRepository<DataEntryEntity, DataEntryId>, Jp
       }
 
     /**
-     * Specification for checking the payload attribute.
+     * Specification for checking the payload attribute. If multiple values are given, one of them must match.
+     * payload.name = ? AND (payload.value = ? OR payload.value = ? OR ...)
      */
-    fun hasDataEntryPayloadAttribute(name: String, value: String): Specification<DataEntryEntity> =
+    fun hasDataEntryPayloadAttribute(name: String, values: List<String>): Specification<DataEntryEntity> =
       Specification { dataEntry, query, builder ->
         query.distinct(true)
         val join = dataEntry.join<DataEntryEntity, Set<PayloadAttribute>>(DataEntryEntity::payloadAttributes.name)
@@ -94,11 +95,15 @@ interface DataEntryRepository : CrudRepository<DataEntryEntity, DataEntryId>, Jp
           join.get<String>(PayloadAttribute::path.name),
           name
         )
-        val valueEquals = builder.equal(
-          join.get<String>(PayloadAttribute::value.name),
-          value
-        )
-        builder.and(pathEquals, valueEquals)
+
+        val valueAnyOf = values.map {
+          builder.equal(
+            join.get<String>(PayloadAttribute::value.name),
+            it
+          )
+        }.let { builder.or(*it.toTypedArray()) }
+
+        builder.and(pathEquals, valueAnyOf)
       }
 
     /**
diff --git a/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/task/TaskRepository.kt b/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/task/TaskRepository.kt
index 2e97a302f..2cb069c33 100644
--- a/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/task/TaskRepository.kt
+++ b/view/jpa/src/main/kotlin/io/holunda/polyflow/view/jpa/task/TaskRepository.kt
@@ -235,9 +235,10 @@ interface TaskRepository : CrudRepository<TaskEntity, String>, JpaSpecificationE
         .or(likeProcessName(pattern))
 
     /**
-     * Specification for checking the payload attribute of a task.
+     * Specification for checking the payload attribute of a task. If multiple values are given, one of them must match.
+     * payload.name = ? AND (payload.value = ? OR payload.value = ? OR ...)
      */
-    fun hasTaskPayloadAttribute(name: String, value: String): Specification<TaskEntity> =
+    fun hasTaskPayloadAttribute(name: String, values: List<String>): Specification<TaskEntity> =
       Specification { task, query, builder ->
         query.distinct(true)
         val join = task.join<TaskEntity, Set<PayloadAttribute>>(TaskEntity::payloadAttributes.name)
@@ -245,11 +246,15 @@ interface TaskRepository : CrudRepository<TaskEntity, String>, JpaSpecificationE
           join.get<String>(PayloadAttribute::path.name),
           name
         )
-        val valueEquals = builder.equal(
-          join.get<String>(PayloadAttribute::value.name),
-          value
-        )
-        builder.and(pathEquals, valueEquals)
+
+        val valueAnyOf = values.map {
+          builder.equal(
+            join.get<String>(PayloadAttribute::value.name),
+            it
+          )
+        }.let { builder.or(*it.toTypedArray()) }
+
+        builder.and(pathEquals, valueAnyOf)
       }
   }
 }


Test patch:
diff --git a/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/JpaPolyflowViewServiceDataEntryITest.kt b/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/JpaPolyflowViewServiceDataEntryITest.kt
index 5c43829a4..525c80690 100644
--- a/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/JpaPolyflowViewServiceDataEntryITest.kt
+++ b/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/JpaPolyflowViewServiceDataEntryITest.kt
@@ -265,7 +265,7 @@ internal class JpaPolyflowViewServiceDataEntryITest {
   fun `should find the entry by filter`() {
     assertResultIsTestEntry1And2(
       jpaPolyflowViewService.query(
-        DataEntriesQuery(filters = listOf("key=value"))
+        DataEntriesQuery(filters = listOf("key=value", "key=value2", "key=value3"))
       )
     )
   }
diff --git a/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/data/DataEntryRepositoryITest.kt b/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/data/DataEntryRepositoryITest.kt
index 1643a2859..e501ebd7f 100644
--- a/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/data/DataEntryRepositoryITest.kt
+++ b/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/data/DataEntryRepositoryITest.kt
@@ -242,13 +242,13 @@ internal class DataEntryRepositoryITest {
   @Test
   fun `should find by filter payload attribute`() {
 
-    val byPayloadFilterByChildKeyNumberValue = dataEntryRepository.findAll(where(hasDataEntryPayloadAttribute("child.key-number", "42")))
+    val byPayloadFilterByChildKeyNumberValue = dataEntryRepository.findAll(where(hasDataEntryPayloadAttribute("child.key-number", listOf("42"))))
     assertThat(byPayloadFilterByChildKeyNumberValue).containsExactlyInAnyOrderElementsOf(listOf(dataEntry, dataEntry2))
 
     val byPayloadFilterByChildKeyValue =
       dataEntryRepository.findAll(
-        where(hasDataEntryPayloadAttribute("child.key", "value"))
-          .and(hasDataEntryPayloadAttribute("id", dataEntry.dataEntryId.entryId))
+        where(hasDataEntryPayloadAttribute("child.key", listOf("value")))
+          .and(hasDataEntryPayloadAttribute("id", listOf(dataEntry.dataEntryId.entryId)))
       )
     assertThat(byPayloadFilterByChildKeyValue).containsExactlyInAnyOrderElementsOf(listOf(dataEntry))
   }
diff --git a/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/task/TaskRepositoryITest.kt b/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/task/TaskRepositoryITest.kt
index 276bca861..272f32f52 100644
--- a/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/task/TaskRepositoryITest.kt
+++ b/view/jpa/src/test/kotlin/io/holunda/polyflow/view/jpa/task/TaskRepositoryITest.kt
@@ -195,11 +195,11 @@ class TaskRepositoryITest {
 
   @Test
   fun `should find by payload attribute`() {
-    val result = taskRepository.findAll(hasTaskPayloadAttribute("complex.child2", "small"))
+    val result = taskRepository.findAll(hasTaskPayloadAttribute("complex.child2", listOf("small")))
     assertThat(result).hasSize(1)
     assertThat(result).containsExactlyInAnyOrder(task1)
 
-    val notfound = taskRepository.findAll(hasTaskPayloadAttribute("complex.child1", "13"))
+    val notfound = taskRepository.findAll(hasTaskPayloadAttribute("complex.child1", listOf("13")))
     assertThat(notfound).isEmpty()
   }