fix(things): 修复聚合时间桶匹配与排序

Author: zhou-hao

jetlinks-components/tdengine-component/src/test/java/org/jetlinks/community/tdengine/things/TDengineSqlSecurityTest.java

@@ -87,15 +87,16 @@ void shouldMapAggregationAliasByPropertyPartition() {
         PropertyAggregation temperatureCount = new PropertyAggregation(
             "temperature", "temperatureCount", Aggregation.COUNT);
 
+        long timestamp = 1_725_000_000_000L;
         TDengineThingDataHelper helper = mock(TDengineThingDataHelper.class);
         when(helper.query(anyString())).thenReturn(Flux.just(
-            TimeSeriesData.of(0, Map.of(
+            TimeSeriesData.of(timestamp, Map.of(
                 "property", "temperature",
                 "__agg_0", 20D,
                 "__agg_1", 20D,
                 "__agg_2", 2L
             )),
-            TimeSeriesData.of(0, Map.of(
+            TimeSeriesData.of(timestamp, Map.of(
                 "property", "humidity",
                 "__agg_0", 80D,
                 "__agg_1", 80D,

jetlinks-components/things-component/src/main/java/org/jetlinks/community/things/data/ThingsDataUtils.java

@@ -15,7 +15,6 @@
  */
 package org.jetlinks.community.things.data;
 
-import org.jetlinks.community.timeseries.utils.TimeSeriesUtils;
 import org.jetlinks.community.Interval;
 import org.jetlinks.community.timeseries.utils.TimeSeriesUtils;
 import org.joda.time.DateTime;
@@ -52,13 +51,13 @@ public static NavigableMap<Long, Map<String, Object>> prepareAggregationData(Agg
     public static NavigableMap<Long, Map<String, Object>> prepareAggregationData(AggregationRequest request,
                                                                                  BiFunction<Long, Interval, Long> timeTruncate,
                                                                                  PropertyAggregation... properties) {
-        NavigableMap<Long, Map<String, Object>> data = new TreeMap<>(Comparator.comparingLong(l -> -l));
+        NavigableMap<Long, Map<String, Object>> data = new TreeMap<>();
         Map<String, Object> valueMap = new HashMap<>();
         for (PropertyAggregation property : properties) {
             valueMap.put(property.getAlias(), property.getDefaultValue());
         }
         if (request.getInterval() == null) {
-            data.put(0L,valueMap);
+            data.put(0L, valueMap);
             return data;
         }
         DateTimeFormatter formatter = DateTimeFormat.forPattern(request.getFormat());

jetlinks-components/things-component/src/test/java/org/jetlinks/community/things/data/ThingsDataUtilsTest.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2026 JetLinks https://www.jetlinks.cn
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.jetlinks.community.things.data;
+
+import org.jetlinks.community.Interval;
+import org.jetlinks.community.timeseries.query.Aggregation;
+import org.junit.jupiter.api.Test;
+
+import java.util.Date;
+import java.util.Map;
+import java.util.NavigableMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+class ThingsDataUtilsTest {
+
+    private static final PropertyAggregation TEMPERATURE =
+        new PropertyAggregation("temperature", "temperatureAvg", Aggregation.AVG);
+
+    @Test
+    void shouldMatchNonZeroTimestampWithoutInterval() {
+        AggregationRequest request = AggregationRequest
+            .builder()
+            .interval(null)
+            .build();
+
+        NavigableMap<Long, Map<String, Object>> prepares =
+            ThingsDataUtils.prepareAggregationData(request, TEMPERATURE);
+
+        assertEquals(0L, prepares.firstKey());
+        assertSame(prepares.get(0L), ThingsDataUtils.findAggregationData(1_725_000_000_000L, prepares));
+    }
+
+    @Test
+    void shouldMatchRawTimestampToPreviousTimeBucket() {
+        long from = 1_725_000_000_000L;
+        AggregationRequest request = AggregationRequest
+            .builder()
+            .interval(Interval.ofHours(1))
+            .format("yyyy-MM-dd HH:mm")
+            .from(new Date(from))
+            .to(new Date(from + 2 * 60 * 60 * 1000L))
+            .build();
+
+        NavigableMap<Long, Map<String, Object>> prepares =
+            ThingsDataUtils.prepareAggregationData(request, (time, interval) -> time, TEMPERATURE);
+
+        long firstBucket = prepares.firstKey();
+        assertSame(
+            prepares.get(firstBucket),
+            ThingsDataUtils.findAggregationData(firstBucket + 30 * 60 * 1000L, prepares)
+        );
+    }
+
+    @Test
+    void shouldPrepareNaturalOrderAndReadNewestFirst() {
+        long from = 1_725_000_000_000L;
+        AggregationRequest request = AggregationRequest
+            .builder()
+            .interval(Interval.ofHours(1))
+            .format("yyyy-MM-dd HH:mm")
+            .from(new Date(from))
+            .to(new Date(from + 2 * 60 * 60 * 1000L))
+            .build();
+
+        NavigableMap<Long, Map<String, Object>> prepares =
+            ThingsDataUtils.prepareAggregationData(request, (time, interval) -> time, TEMPERATURE);
+
+        assertEquals(prepares.lastKey(), prepares.descendingMap().firstKey());
+    }
+}

jetlinks-components/timescaledb-component/src/main/java/org/jetlinks/community/timescaledb/thing/TimescaleDBColumnModeQueryOperations.java

@@ -177,7 +177,7 @@ static Flux<AggregationData> doAggregation0(DatabaseOperator database,
                         }
                     });
             })
-            .thenMany(Flux.fromIterable(prepares.values()))
+            .thenMany(Flux.fromIterable(prepares.descendingMap().values()))
             .map(AggregationData::of)
             .take((long) request.getLimit() * context.getProperties().length)
             .contextWrite(ctx -> ctx.put(Logger.class, log));

jetlinks-components/timescaledb-component/src/main/java/org/jetlinks/community/timescaledb/thing/TimescaleDBRowModeQueryOperations.java

@@ -191,7 +191,7 @@ protected Flux<AggregationData> doAggregation(String metric,
                         }
                     });
             })
-            .thenMany(Flux.fromIterable(prepares.values()))
+            .thenMany(Flux.fromIterable(prepares.descendingMap().values()))
             .map(AggregationData::of)
             .take((long) request.getLimit() * propertyId.size())
             .contextWrite(ctx -> ctx.put(Logger.class, log));