diff --git a/model/src/main/java/org/openremote/model/datapoint/query/AssetDatapointIntervalQuery.java b/model/src/main/java/org/openremote/model/datapoint/query/AssetDatapointIntervalQuery.java index e792c4c08d..ef88b93d08 100644 --- a/model/src/main/java/org/openremote/model/datapoint/query/AssetDatapointIntervalQuery.java +++ b/model/src/main/java/org/openremote/model/datapoint/query/AssetDatapointIntervalQuery.java @@ -47,8 +47,7 @@ public class AssetDatapointIntervalQuery extends AssetDatapointQuery { description = "When true, returns empty buckets across the requested range using time-bucket gap " + "filling; their aggregate is generally null. When false, only buckets containing datapoints are " - + "returned. MODE currently returns only buckets containing datapoints regardless of this flag. " - + "Defaults to false.", + + "returned. Empty buckets count toward the datapoint query limit. Defaults to false.", example = "true") public boolean gapFill; @@ -157,30 +156,18 @@ protected String getSQLQueryForNumbers(String tableName, String function) { + "WHERE ENTITY_ID = ? and ATTRIBUTE_NAME = ? and TIMESTAMP >= ? and TIMESTAMP <= ? " + "GROUP BY x ORDER by x ASC"; case MODE: - return "WITH bucketed AS ( " - + " SELECT " + // Relies on PostgreSQL MODE retaining the first tied value in sort order (ASC: smaller). + return "SELECT " + function + "(cast(? as interval), timestamp) AS x, " - + " cast(" + + "MODE() WITHIN GROUP (ORDER BY cast(" + NULLABLE_VALUE_SQL - + " as numeric) AS num_value " - + " FROM " + + " as numeric) ASC) AS mode_value " + + "FROM " + tableName + " " - + " WHERE ENTITY_ID = ? and ATTRIBUTE_NAME = ? and TIMESTAMP >= ? and TIMESTAMP <= ? " - + ") " - + "SELECT d.x, ( " - + " SELECT num_value FROM ( " - + " SELECT num_value, COUNT(*) AS freq " - + " FROM bucketed b2 " - + " WHERE b2.x = d.x AND num_value IS NOT NULL " - + " GROUP BY num_value " - + " ORDER BY freq DESC, num_value " - + " LIMIT 1 " - + " ) AS mode_sub " - + ") AS mode_value " - + "FROM (SELECT DISTINCT x FROM bucketed) d " - + "ORDER BY d.x ASC"; + + "WHERE ENTITY_ID = ? and ATTRIBUTE_NAME = ? and TIMESTAMP >= ? and TIMESTAMP <= ? " + + "GROUP BY x ORDER by x ASC"; case MEDIAN: return "SELECT " + function @@ -234,30 +221,18 @@ protected String getSQLQueryForBooleans(String tableName, String function) { + "WHERE ENTITY_ID = ? and ATTRIBUTE_NAME = ? and TIMESTAMP >= ? and TIMESTAMP <= ? " + "GROUP BY x ORDER by x ASC"; case MODE: - return "WITH bucketed AS ( " - + " SELECT " + // Relies on PostgreSQL MODE retaining the first tied value in sort order (DESC: true). + return "SELECT " + function + "(cast(? as interval), timestamp) AS x, " - + " cast(" + + "MODE() WITHIN GROUP (ORDER BY cast(" + NULLABLE_VALUE_SQL - + " as text)::boolean AS bool_value " - + " FROM " + + " as text)::boolean DESC)::integer AS mode_value " + + "FROM " + tableName + " " - + " WHERE ENTITY_ID = ? and ATTRIBUTE_NAME = ? and TIMESTAMP >= ? and TIMESTAMP <= ? " - + ") " - + "SELECT d.x, ( " - + " SELECT bool_value::integer FROM ( " - + " SELECT bool_value, COUNT(*) AS freq " - + " FROM bucketed b2 " - + " WHERE b2.x = d.x AND bool_value IS NOT NULL " - + " GROUP BY bool_value " - + " ORDER BY freq DESC, bool_value DESC " - + " LIMIT 1 " - + " ) AS mode_sub " - + ") AS mode_value " - + "FROM (SELECT DISTINCT x FROM bucketed) d " - + "ORDER BY d.x ASC"; + + "WHERE ENTITY_ID = ? and ATTRIBUTE_NAME = ? and TIMESTAMP >= ? and TIMESTAMP <= ? " + + "GROUP BY x ORDER by x ASC"; case MEDIAN: return "SELECT " + function diff --git a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointNullQueryTest.groovy b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointNullQueryTest.groovy index dfeacc9d84..164e4f86ee 100644 --- a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointNullQueryTest.groovy +++ b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointNullQueryTest.groovy @@ -24,6 +24,7 @@ import org.openremote.manager.datapoint.AssetPredictedDatapointService import org.openremote.manager.setup.SetupService import org.openremote.model.asset.impl.ThingAsset import org.openremote.model.attribute.Attribute +import org.openremote.model.datapoint.DatapointQueryTooLargeException import org.openremote.model.datapoint.ValueDatapoint import org.openremote.model.datapoint.query.AssetDatapointAllQuery import org.openremote.model.datapoint.query.AssetDatapointIntervalQuery @@ -277,8 +278,8 @@ class AssetDatapointNullQueryTest extends Specification implements ManagerContai new ValueDatapoint<>(bucketStart + bucket * 60_000L + index * 1_000L, value) } } - datapoints.add(new ValueDatapoint<>(bucketStart - 1_000L, high)) - datapoints.add(new ValueDatapoint<>(bucketStart + 360_000L, low)) + datapoints.add(new ValueDatapoint<>(bucketStart - 61_000L, high)) + datapoints.add(new ValueDatapoint<>(bucketStart + 420_000L, low)) service.upsertValues(asset.id, "value", datapoints) when: "the six-minute range is queried for its most frequent available values" @@ -287,10 +288,32 @@ class AssetDatapointNullQueryTest extends Specification implements ManagerContai "1 minute", Formula.MODE, gapFill)) then: "nulls cannot win and existing numeric and boolean tie-breaking is preserved" - // Preserve existing MODE behavior: empty buckets are omitted for either gapFill setting. - result.collect { it.timestamp } == [0L, 60_000L, 180_000L, 240_000L, 300_000L] - .collect { bucketStart + it } - result.collect { it.value } == [lowResult, null, tieResult, 0D, highResult] + result.collect { it.timestamp } == (gapFill + ? [0L, 60_000L, 120_000L, 180_000L, 240_000L, 300_000L] + : [0L, 60_000L, 180_000L, 240_000L, 300_000L]).collect { bucketStart + it } + result.collect { it.value } == (gapFill + ? [lowResult, null, null, tieResult, 0D, highResult] + : [lowResult, null, tieResult, 0D, highResult]) + + when: "the range includes empty leading and trailing buckets" + def extended = service.queryDatapoints(asset.id, "value", + new AssetDatapointIntervalQuery(bucketStart - 60_000L, bucketStart + 419_999L, + "1 minute", Formula.MODE, gapFill)) + + then: "gap filling covers the requested range rather than just the observed timestamps" + extended.collect { [it.timestamp, it.value] } == (gapFill + ? [[bucketStart - 60_000L, null]] + result.collect { + [it.timestamp, it.value] + } + [[bucketStart + 360_000L, null]] + : result.collect { [it.timestamp, it.value] }) + + when: "only the empty bucket is queried" + def empty = service.queryDatapoints(asset.id, "value", + new AssetDatapointIntervalQuery(bucketStart + 120_000L, bucketStart + 179_999L, + "1 minute", Formula.MODE, gapFill)) + + then: "an entirely empty range also follows the gap-fill setting" + empty.collect { [it.timestamp, it.value] } == (gapFill ? [[bucketStart + 120_000L, null]] : []) when: "only the entirely unavailable bucket is queried" def unavailable = service.queryDatapoints(asset.id, "value", @@ -310,6 +333,43 @@ class AssetDatapointNullQueryTest extends Specification implements ManagerContai } } + def "MODE counts gap-filled buckets toward the query limit for #storage #valueType.name"() { + given: "a sparse range with more buckets than the query limit" + asset = assetStorageService.merge(new ThingAsset("Mode query limit test") + .setRealm(realm).addOrReplaceAttributes(new Attribute<>("value", valueType))) + def service = storage == "historical" ? historicalService : predictedService + def originalLimit = service.maxAmountOfQueryPoints + def bucketStart = Instant.ofEpochMilli(start).truncatedTo(ChronoUnit.MINUTES).toEpochMilli() + def value = valueType == ValueType.BOOLEAN ? true : 1G + service.upsertValues(asset.id, "value", [ + new ValueDatapoint<>(bucketStart, value), + new ValueDatapoint<>(bucketStart + 180_000L, value) + ]) + service.maxAmountOfQueryPoints = 2 + + when: "only populated buckets are requested" + def result = service.queryDatapoints(asset.id, "value", + new AssetDatapointIntervalQuery(bucketStart, bucketStart + 239_999L, + "1 minute", Formula.MODE, false)) + + then: "the two populated buckets fit within the limit" + result.collect { it.timestamp } == [bucketStart, bucketStart + 180_000L] + + when: "gap filling increases the result to four buckets" + service.queryDatapoints(asset.id, "value", + new AssetDatapointIntervalQuery(bucketStart, bucketStart + 239_999L, + "1 minute", Formula.MODE, true)) + + then: "the generated empty buckets also count toward the limit" + thrown(DatapointQueryTooLargeException) + + cleanup: + service.maxAmountOfQueryPoints = originalLimit + + where: + [storage, valueType] << [["historical", "predicted"], [ValueType.NUMBER, ValueType.BIG_NUMBER, ValueType.BOOLEAN]].combinations() + } + def "DIFFERENCE preserves unavailable values for #storage #valueType.name with gapFill=#gapFill"() { given: "last values, explicit unavailability, recovery, and empty minute buckets" asset = assetStorageService.merge(new ThingAsset("Difference null query test") diff --git a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointQueryTest.groovy b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointQueryTest.groovy index 79cd81f6ec..5f43f910a5 100644 --- a/test/src/test/groovy/org/openremote/test/assets/AssetDatapointQueryTest.groovy +++ b/test/src/test/groovy/org/openremote/test/assets/AssetDatapointQueryTest.groovy @@ -444,19 +444,21 @@ class AssetDatapointQueryTest extends Specification implements ManagerContainerT ) then: "MODE datapoints should be correct" - assert modeDatapoints1.size() == 6 + assert modeDatapoints1.size() == 7 // Include the empty bucket at the inclusive end timestamp. assert modeDatapoints1[0].timestamp == time30.toEpochMilli() assert modeDatapoints1[1].timestamp == time30.plus(5, ChronoUnit.MINUTES).toEpochMilli() assert modeDatapoints1[2].timestamp == time30.plus(10, ChronoUnit.MINUTES).toEpochMilli() assert modeDatapoints1[3].timestamp == time30.plus(15, ChronoUnit.MINUTES).toEpochMilli() assert modeDatapoints1[4].timestamp == time30.plus(20, ChronoUnit.MINUTES).toEpochMilli() assert modeDatapoints1[5].timestamp == time30.plus(25, ChronoUnit.MINUTES).toEpochMilli() + assert modeDatapoints1[6].timestamp == timeNow.toEpochMilli() assert modeDatapoints1[0].value == 10 assert modeDatapoints1[1].value == 15 assert modeDatapoints1[2].value == 90 assert modeDatapoints1[3].value == 20 assert modeDatapoints1[4].value == 25 assert modeDatapoints1[5].value == 30 + assert modeDatapoints1[6].value == null when: "requesting interval datapoints using MEDIAN" def medianDatapoints1 = assetDatapointService.queryDatapoints( @@ -489,9 +491,11 @@ class AssetDatapointQueryTest extends Specification implements ManagerContainerT ) then: "MODE datapoints should return 90, because it is most common within the interval" - assert modeDatapoints2.size() == 1 + assert modeDatapoints2.size() == 2 // Include the empty bucket at the inclusive end timestamp. assert modeDatapoints2[0].timestamp == time60.toEpochMilli() assert modeDatapoints2[0].value == 90 + assert modeDatapoints2[1].timestamp == timeNow.toEpochMilli() + assert modeDatapoints2[1].value == null } def "All query should return the correct data when the maximum is respected"() {