fix(tdengine): 修复多属性聚合结果串值

Author: zhou-hao

jetlinks-components/tdengine-component/src/main/java/org/jetlinks/community/tdengine/things/TDengineRowModeQueryOperations.java

@@ -25,15 +25,14 @@
 import org.jetlinks.community.things.data.AggregationRequest;
 import org.jetlinks.community.things.data.PropertyAggregation;
 import org.jetlinks.community.things.data.ThingPropertyDetail;
-import org.jetlinks.community.things.data.ThingsDataConstants;
+import org.jetlinks.community.things.data.ThingsDataUtils;
 import org.jetlinks.community.things.data.operations.DataSettings;
 import org.jetlinks.community.things.data.operations.MetricBuilder;
 import org.jetlinks.community.things.data.operations.RowModeQueryOperationsBase;
 import org.jetlinks.community.timeseries.TimeSeriesData;
 import org.jetlinks.community.timeseries.query.Aggregation;
 import org.jetlinks.community.timeseries.query.AggregationData;
 import org.jetlinks.community.utils.SqlSecurityUtils;
-import org.jetlinks.reactor.ql.utils.CastUtils;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -44,6 +43,7 @@
 import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 class TDengineRowModeQueryOperations extends RowModeQueryOperationsBase {
 
@@ -138,36 +138,32 @@ protected Flux<AggregationData> doAggregation(String metric,
                 .take(request.getLimit())
                 ;
         }
+        NavigableMap<Long, Map<String, Object>> prepares =
+            ThingsDataUtils.prepareAggregationData(request, properties);
+        Map<String, List<PropertyAggregation>> propertyAgg = Arrays
+            .stream(properties)
+            .collect(Collectors.groupingBy(PropertyAggregation::getProperty));
         return helper
             .query(dataSql)
-            .map(timeSeriesData -> {
-                long ts = timeSeriesData.getTimestamp();
-                Map<String, Object> newData = timeSeriesData.getData();
-                newData.put("time", formatter.format(LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneId.systemDefault())));
-                newData.put("_time", ts);
-                return newData;
-            })
-            .groupBy(data -> (String) data.get("time"), Integer.MAX_VALUE)
-            .flatMap(group -> group
-                .reduceWith(HashMap::new, (a, b) -> {
-                    a.putAll(b);
-                    return a;
-                })
-                .map(map -> {
-                    Map<String, Object> newResult = new HashMap<>();
-                    for (PropertyAggregation property : properties) {
-                        String alias = property.getAlias();
-                        String key = aliases.get(alias);
-                        newResult.put(alias, Optional.ofNullable(map.get(key)).orElse(property.getDefaultValue()));
+            .doOnNext(data -> {
+                long timestamp = data.getTimestamp();
+                Map<String, Object> prepare = ThingsDataUtils.findAggregationData(timestamp, prepares);
+                if (prepare != null) {
+                    Object propertyValue = data.getData().get("property");
+                    List<PropertyAggregation> proAggs = propertyValue == null
+                        ? null
+                        : propertyAgg.get(propertyValue.toString());
+                    // 每个 partition 行都会计算全部投影，只消费当前 property 对应的聚合列。
+                    if (proAggs != null) {
+                        for (PropertyAggregation proAgg : proAggs) {
+                            String alias = proAgg.getAlias();
+                            prepare.put(alias, data.get(aliases.get(alias)).orElse(proAgg.getDefaultValue()));
+                        }
                     }
-                    newResult.put("time", group.key());
-                    newResult.put("_time", map.getOrDefault("_time", new Date()));
-                    return AggregationData.of(newResult);
-                }))
-            .sort(Comparator
-                      .<AggregationData, Date>comparing(data -> CastUtils.castDate(data.values().get("_time")))
-                      .reversed())
-            .doOnNext(data -> data.values().remove("_time"))
+                }
+            })
+            .thenMany(Flux.fromIterable(prepares.descendingMap().values()))
+            .map(AggregationData::of)
             .take(request.getLimit());
     }
 

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

@@ -15,12 +15,25 @@
  */
 package org.jetlinks.community.tdengine.things;
 
+import org.jetlinks.core.things.ThingMetadata;
+import org.jetlinks.community.things.data.AggregationRequest;
 import org.jetlinks.community.things.data.PropertyAggregation;
+import org.jetlinks.community.things.data.operations.DataSettings;
+import org.jetlinks.community.things.data.operations.MetricBuilder;
+import org.jetlinks.community.timeseries.TimeSeriesData;
 import org.jetlinks.community.timeseries.query.Aggregation;
+import org.jetlinks.community.timeseries.query.AggregationData;
 import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+
+import java.util.List;
+import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 class TDengineSqlSecurityTest {
 
@@ -64,4 +77,68 @@ void shouldEscapePropertyAndUseInternalAliasInColumnMode() {
         assertEquals("max(`temperature``), current_user --`) `__agg_1`", sql);
         assertFalse(sql.contains(aggregation.getAlias()));
     }
+
+    @Test
+    void shouldMapAggregationAliasByPropertyPartition() {
+        PropertyAggregation temperatureAvg = new PropertyAggregation(
+            "temperature", "temperatureAvg", Aggregation.AVG);
+        PropertyAggregation humidityMax = new PropertyAggregation(
+            "humidity", "humidityMax", Aggregation.MAX);
+        PropertyAggregation temperatureCount = new PropertyAggregation(
+            "temperature", "temperatureCount", Aggregation.COUNT);
+
+        TDengineThingDataHelper helper = mock(TDengineThingDataHelper.class);
+        when(helper.query(anyString())).thenReturn(Flux.just(
+            TimeSeriesData.of(0, Map.of(
+                "property", "temperature",
+                "__agg_0", 20D,
+                "__agg_1", 20D,
+                "__agg_2", 2L
+            )),
+            TimeSeriesData.of(0, Map.of(
+                "property", "humidity",
+                "__agg_0", 80D,
+                "__agg_1", 80D,
+                "__agg_2", 3L
+            ))
+        ));
+
+        AggregationRequest request = AggregationRequest
+            .builder()
+            .interval(null)
+            .limit(1)
+            .build();
+
+        List<AggregationData> result = new TestOperations(helper)
+            .aggregate(request, temperatureAvg, humidityMax, temperatureCount)
+            .collectList()
+            .block();
+
+        assertEquals(1, result.size(), String.valueOf(result));
+        assertEquals(
+            Map.of(
+                "temperatureAvg", 20D,
+                "humidityMax", 80D,
+                "temperatureCount", 2L
+            ),
+            result.get(0).asMap()
+        );
+    }
+
+    private static class TestOperations extends TDengineRowModeQueryOperations {
+
+        TestOperations(TDengineThingDataHelper helper) {
+            super("device", "product", null,
+                  MetricBuilder.DEFAULT, new DataSettings(), null, helper);
+        }
+
+        Flux<AggregationData> aggregate(AggregationRequest request,
+                                        PropertyAggregation... properties) {
+            return doAggregation(
+                "device_properties_product",
+                request,
+                new AggregationContext(mock(ThingMetadata.class), properties)
+            );
+        }
+    }
 }