Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
ffe679e
Handle JSON null in numeric All datapoint queries
ebariaux Oct 9, 2026
3b13522
Handle JSON null in nearest datapoint queries
ebariaux Oct 9, 2026
446f09d
Handle JSON null in standard numeric datapoint aggregates
ebariaux Oct 9, 2026
923b46d
Fix null handling in boolean datapoint All queries
ebariaux Oct 9, 2026
fedca78
Fix null handling in boolean interval aggregates
ebariaux Oct 9, 2026
3133168
Fix null handling in numeric and boolean MODE queries
ebariaux Oct 9, 2026
70412b4
Fix: honor gap filling in MODE datapoint queries
ebariaux Oct 9, 2026
69b56f6
Fix: preserve unavailable values in DIFFERENCE queries
ebariaux Oct 9, 2026
e38cb9f
Fix null handling in numeric and boolean LTTB queries
ebariaux Oct 9, 2026
16758ed
Clarify some doc elements
ebariaux Oct 9, 2026
0b44517
Merge branch 'main' into bugfix/queries-json-null
ebariaux Oct 20, 2026
ee96bed
Merge branch 'main' into bugfix/queries-json-null
ebariaux Oct 21, 2026
e8a6756
Merge branch 'main' into bugfix/queries-json-null
ebariaux Oct 22, 2026
b350365
Fix pseudo-clock changing the JVM default time zone
ebariaux Oct 23, 2026
5a8735f
Separate MODE gap filling from the JSON-null fixes
ebariaux Oct 23, 2026
05dc7c4
Fix MODE queries to include empty buckets when gapFill is enabled
ebariaux Oct 23, 2026
707bd79
Document PostgreSQL MODE tie-breaking dependency
ebariaux Oct 23, 2026
212d9dd
Merge branch 'main' into bugfix/mode-query-gap-fill
wborn Oct 26, 2026
9c56c42
Linting
ebariaux Oct 26, 2026
a6281a1
Merge main into bugfix/mode-query-gap-fill
ebariaux Oct 27, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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"
Expand All @@ -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",
Expand All @@ -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")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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"() {
Expand Down
Loading