first draft for the queries with lastN parameter

stellio-hub · May 15, 2024 · 2b352cc · 2b352cc
1 parent 0023f9f
commit 2b352cc
Show file tree

Hide file tree

Showing 3 changed files with 49 additions and 10 deletions.
diff --git a/search-service/config/detekt/baseline.xml b/search-service/config/detekt/baseline.xml
@@ -6,8 +6,10 @@
     <ID>ClassNaming:V0_29__JsonLd_migration.kt$V0_29__JsonLd_migration : BaseJavaMigration</ID>
     <ID>ComplexCondition:EntitiesQueryUtils.kt$geoQuery == null &amp;&amp; q.isNullOrEmpty() &amp;&amp; typeSelection.isNullOrEmpty() &amp;&amp; attrs.isEmpty()</ID>
     <ID>ComplexCondition:EntityPayloadService.kt$EntityPayloadService$it &amp;&amp; !inverse || !it &amp;&amp; inverse</ID>
+    <ID>CyclomaticComplexMethod:AttributeInstanceService.kt$AttributeInstanceService$suspend fun search( temporalEntitiesQuery: TemporalEntitiesQuery, temporalEntityAttributes: List&lt;TemporalEntityAttribute&gt;, origin: ZonedDateTime? = null ): Either&lt;APIException, List&lt;AttributeInstanceResult&gt;&gt;</ID>
     <ID>Filename:V0_29__JsonLd_migration.kt$db.migration.V0_29__JsonLd_migration.kt</ID>
     <ID>LongMethod:AttributeInstanceService.kt$AttributeInstanceService$@Transactional suspend fun create(attributeInstance: AttributeInstance): Either&lt;APIException, Unit&gt;</ID>
+    <ID>LongMethod:AttributeInstanceService.kt$AttributeInstanceService$suspend fun search( temporalEntitiesQuery: TemporalEntitiesQuery, temporalEntityAttributes: List&lt;TemporalEntityAttribute&gt;, origin: ZonedDateTime? = null ): Either&lt;APIException, List&lt;AttributeInstanceResult&gt;&gt;</ID>
     <ID>LongMethod:EntityAccessControlHandler.kt$EntityAccessControlHandler$@PostMapping("/{subjectId}/attrs", consumes = [MediaType.APPLICATION_JSON_VALUE, JSON_LD_CONTENT_TYPE]) suspend fun addRightsOnEntities( @RequestHeader httpHeaders: HttpHeaders, @PathVariable subjectId: String, @RequestBody requestBody: Mono&lt;String&gt; ): ResponseEntity&lt;*&gt;</ID>
     <ID>LongMethod:EntityOperationHandlerTests.kt$EntityOperationHandlerTests$@Test fun `create batch entity should return a 207 when some entities already exist`()</ID>
     <ID>LongMethod:PatchAttributeTests.kt$PatchAttributeTests.Companion$@JvmStatic fun mergePatchProvider(): Stream&lt;Arguments&gt;</ID>
@@ -26,6 +28,8 @@
     <ID>LongParameterList:TemporalEntityAttributeService.kt$TemporalEntityAttributeService$( tea: TemporalEntityAttribute, attributeName: ExpandedTerm, attributeMetadata: AttributeMetadata, mergedAt: ZonedDateTime, observedAt: ZonedDateTime?, attributePayload: ExpandedAttributeInstance, sub: Sub? )</ID>
     <ID>LongParameterList:TemporalEntityAttributeService.kt$TemporalEntityAttributeService$( temporalEntityAttribute: TemporalEntityAttribute, ngsiLdAttribute: NgsiLdAttribute, attributeMetadata: AttributeMetadata, createdAt: ZonedDateTime, attributePayload: ExpandedAttributeInstance, sub: Sub? )</ID>
     <ID>LongParameterList:V0_29__JsonLd_migration.kt$V0_29__JsonLd_migration$( entityId: URI, attributeName: ExpandedTerm, datasetId: URI?, attributePayload: ExpandedAttributeInstance, ngsiLdAttributeInstance: NgsiLdAttributeInstance, defaultCreatedAt: ZonedDateTime )</ID>
+    <ID>MaxLineLength:AttributeInstanceServiceTests.kt$AttributeInstanceServiceTests$fun</ID>
+    <ID>MaximumLineLength:AttributeInstanceServiceTests.kt$AttributeInstanceServiceTests$ </ID>
     <ID>NestedBlockDepth:V0_29__JsonLd_migration.kt$V0_29__JsonLd_migration$override fun migrate(context: Context)</ID>
     <ID>ReturnCount:EntitiesQueryUtils.kt$fun buildTemporalQuery( params: MultiValueMap&lt;String, String&gt;, inQueryEntities: Boolean = false, withAggregatedValues: Boolean = false ): Either&lt;APIException, TemporalQuery&gt;</ID>
     <ID>SwallowedException:EntitiesQueryUtils.kt$e: IllegalArgumentException</ID>

diff --git a/search-service/src/main/kotlin/com/egm/stellio/search/service/AttributeInstanceService.kt b/search-service/src/main/kotlin/com/egm/stellio/search/service/AttributeInstanceService.kt
@@ -126,9 +126,11 @@ class AttributeInstanceService(
 
         sqlQueryBuilder.append(composeSearchSelectStatement(temporalQuery, temporalEntityAttributes, origin))
 
-        if (!temporalEntitiesQuery.withTemporalValues && !temporalEntitiesQuery.withAggregatedValues)
+        if (!temporalEntitiesQuery.withTemporalValues && !temporalEntitiesQuery.withAggregatedValues &&
+            temporalQuery.aggrMethods == null
+        ) {
             sqlQueryBuilder.append(", payload")
-
+        }
         if (temporalQuery.timeproperty == AttributeInstance.TemporalProperty.OBSERVED_AT)
             sqlQueryBuilder.append(
                 """
@@ -155,12 +157,19 @@ class AttributeInstanceService(
         }
 
         if (temporalEntitiesQuery.isAggregatedWithDefinedDuration())
-            sqlQueryBuilder.append(" GROUP BY temporal_entity_attribute, origin")
+            sqlQueryBuilder.append(" GROUP BY temporal_entity_attribute, time")
         else if (temporalEntitiesQuery.withAggregatedValues)
             sqlQueryBuilder.append(" GROUP BY temporal_entity_attribute")
-        else if (temporalQuery.lastN != null)
+        if (temporalQuery.lastN != null && !temporalEntitiesQuery.isAggregatedWithDefinedDuration() &&
+            !temporalEntitiesQuery.withAggregatedValues
+        ) {
             // in order to get last instances, need to order by time desc
             // final ascending ordering of instances is done in query service
+            sqlQueryBuilder.append(
+                " GROUP BY temporal_entity_attribute, time, measured_value, payload " +
+                    "ORDER BY time DESC LIMIT ${temporalQuery.lastN}"
+            )
+        } else if (temporalQuery.lastN != null)
             sqlQueryBuilder.append(" ORDER BY time DESC LIMIT ${temporalQuery.lastN}")
 
         val finalTemporalQuery = composeFinalTemporalQuery(temporalEntityAttributes, sqlQueryBuilder.toString())
@@ -211,11 +220,11 @@ class AttributeInstanceService(
                 val computedOrigin = origin ?: temporalQuery.timeAt
                 """
                 SELECT temporal_entity_attribute,
-                    public.time_bucket('$aggrPeriodDuration', time, TIMESTAMPTZ '${computedOrigin!!}') as origin,
+                    public.time_bucket('$aggrPeriodDuration', time, TIMESTAMPTZ '${computedOrigin!!}') as time,
                     $allAggregates
                 """.trimIndent()
             } else
-                "SELECT temporal_entity_attribute, min(time) as origin, max(time) as endTime, $allAggregates "
+                "SELECT temporal_entity_attribute, min(time) as time, max(time) as endTime, $allAggregates "
         }
         else -> {
             val valueColumn = when (temporalEntityAttributes[0].attributeValueType) {
@@ -269,12 +278,14 @@ class AttributeInstanceService(
         row: Map<String, Any>,
         temporalEntitiesQuery: TemporalEntitiesQuery
     ): AttributeInstanceResult =
-        if (temporalEntitiesQuery.withAggregatedValues) {
-            val startDateTime = toZonedDateTime(row["origin"])
+        if (temporalEntitiesQuery.withAggregatedValues || temporalEntitiesQuery.temporalQuery.aggrMethods != null) {
+            val startDateTime = toZonedDateTime(row["time"])
             val endDateTime =
-                if (!temporalEntitiesQuery.isAggregatedWithDefinedDuration())
+                if (!temporalEntitiesQuery.isAggregatedWithDefinedDuration() &&
+                    temporalEntitiesQuery.temporalQuery.aggrPeriodDuration == null
+                ) {
                     toZonedDateTime(row["endTime"])
-                else
+                } else
                     startDateTime.plus(temporalEntitiesQuery.computeAggrPeriodDuration())
             // in a row, there is the result for each requested aggregation method
             val values = temporalEntitiesQuery.temporalQuery.aggrMethods!!.map {

diff --git a/...h-service/src/test/kotlin/com/egm/stellio/search/service/AttributeInstanceServiceTests.kt b/...h-service/src/test/kotlin/com/egm/stellio/search/service/AttributeInstanceServiceTests.kt
@@ -365,6 +365,30 @@ class AttributeInstanceServiceTests : WithTimescaleContainer, WithKafkaContainer
             }
     }
 
+    @Test
+    fun `it should only return the last n instances with aggregated values and aggregation period in the query`() = runTest {
+        (1..10).forEach { _ ->
+            val attributeInstance = gimmeNumericPropertyAttributeInstance(incomingTemporalEntityAttribute.id)
+                .copy(measuredValue = 1.0)
+            attributeInstanceService.create(attributeInstance)
+        }
+
+        val temporalEntitiesQuery = gimmeTemporalEntitiesQuery(
+            TemporalQuery(
+                timerel = TemporalQuery.Timerel.BEFORE,
+                timeAt = now.plusHours(1),
+                aggrPeriodDuration = "PT1S",
+                aggrMethods = listOf(TemporalQuery.Aggregate.SUM),
+                lastN = 5
+            )
+        )
+        attributeInstanceService.search(temporalEntitiesQuery, incomingTemporalEntityAttribute)
+            .shouldSucceedWith {
+                assertThat(it)
+                    .hasSize(5)
+            }
+    }
+
     @Test
     fun `it should only retrieve the temporal evolution of the provided temporal entity attribute`() = runTest {
         val temporalEntityAttribute2 = TemporalEntityAttribute(