feat(timescaledb): 支持配置TimescaleDB函数所在的schema

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

@@ -39,6 +39,11 @@ public class TimescaleDBProperties {
     //数据库的schema
     private String schema = "public";
 
+    /**
+     * TimescaleDB超表函数所在的位置
+     */
+    private String functionSchema = "public";
+
     /**
      * 写入缓冲区配置
      *

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

@@ -40,15 +40,15 @@ public static String getTableName(String name) {
         return ThingsDatabaseUtils.createTableName(name);
     }
 
-    public static NativeSelectColumn createTimeGroupColumn(long startWith, Interval interval) {
+    public static NativeSelectColumn createTimeGroupColumn(long startWith, Interval interval, String functionSchema) {
 
         String unit = interval.getNumber().intValue() + " " + interval
             .getUnit()
             .name()
             .toLowerCase();
 
         return NativeSelectColumn
-            .of("time_bucket('" + unit + "',timestamp)");
+            .of(functionSchema + ".time_bucket('" + unit + "',timestamp)");
     }
 
     public static TimeSeriesData convertToTimeSeriesData(Record record) {

jetlinks-components/timescaledb-component/src/main/java/org/jetlinks/community/timescaledb/configuration/TimescaleDBTimeSeriesConfiguration.java

@@ -16,6 +16,7 @@
 package org.jetlinks.community.timescaledb.configuration;
 
 import org.jetlinks.community.timescaledb.TimescaleDBOperations;
+import org.jetlinks.community.timescaledb.TimescaleDBProperties;
 import org.jetlinks.community.timescaledb.timeseries.TimescaleDBTimeSeriesManager;
 import org.jetlinks.community.timescaledb.timeseries.TimescaleDBTimeSeriesProperties;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
@@ -35,8 +36,9 @@ public class TimescaleDBTimeSeriesConfiguration {
     @Bean
     @Primary
     public TimescaleDBTimeSeriesManager timescaleDBTimeSeriesManager(TimescaleDBOperations operations,
-                                                                     TimescaleDBTimeSeriesProperties properties) {
-        return new TimescaleDBTimeSeriesManager(properties, operations);
+                                                                     TimescaleDBTimeSeriesProperties properties,
+                                                                     TimescaleDBProperties timescaleDBProperties) {
+        return new TimescaleDBTimeSeriesManager(properties, operations,timescaleDBProperties);
     }
 
 

jetlinks-components/timescaledb-component/src/main/java/org/jetlinks/community/timescaledb/metadata/FunctionSchema.java

@@ -0,0 +1,31 @@
+package org.jetlinks.community.timescaledb.metadata;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.hswebframework.ezorm.core.FeatureId;
+import org.hswebframework.ezorm.core.FeatureType;
+import org.hswebframework.ezorm.core.meta.Feature;
+
+@AllArgsConstructor(staticName = "of")
+@Getter
+public class FunctionSchema implements Feature, FeatureType {
+
+    public static final FeatureId<FunctionSchema> ID = FeatureId.of("FunctionSchema");
+
+    private final String functionSchema;
+
+    @Override
+    public String getId() {
+        return ID.getId();
+    }
+
+    @Override
+    public String getName() {
+        return "FunctionSchema";
+    }
+
+    @Override
+    public FeatureType getType() {
+        return this;
+    }
+}

jetlinks-components/timescaledb-component/src/main/java/org/jetlinks/community/timescaledb/metadata/TimescaleDBCreateTableSqlBuilder.java

@@ -23,16 +23,11 @@
 
 public class TimescaleDBCreateTableSqlBuilder extends CommonCreateTableSqlBuilder {
 
-    private String schema;
-
-    public TimescaleDBCreateTableSqlBuilder(String schema) {
-        this.schema = schema;
-    }
-
     @Override
     public SqlRequest build(RDBTableMetadata table) {
         DefaultBatchSqlRequest sqlRequest = (DefaultBatchSqlRequest) super.build(table);
 
+
         table.getFeature(CreateHypertable.ID)
              .ifPresent(createHypertable -> sqlRequest.addBatch(createCreateHypertableSQL(table, createHypertable)));
 
@@ -47,9 +42,11 @@ private SqlRequest createCreateRetentionPolicySQL(RDBTableMetadata table, Create
 
         String interval = createHypertable.getInterval().getNumber().intValue() + " "
             + createHypertable.getInterval().getUnit().name().toLowerCase();
+        String functionSchema = table.getFeatureNow(FunctionSchema.ID)
+                                     .getFunctionSchema();
 
         return SqlRequests.of(
-            "SELECT "+ schema +".add_retention_policy( ? , INTERVAL '" + interval + "')",
+            "SELECT " + functionSchema + ".add_retention_policy( ? , INTERVAL '" + interval + "')",
             table.getFullName()
         );
     }
@@ -58,9 +55,10 @@ private SqlRequest createCreateHypertableSQL(RDBTableMetadata table, CreateHyper
 
         String interval = createHypertable.getChunkTimeInterval().getNumber().intValue() + " "
             + createHypertable.getChunkTimeInterval().getUnit().name().toLowerCase();
-
+        String functionSchema = table.getFeatureNow(FunctionSchema.ID)
+                                     .getFunctionSchema();
         return SqlRequests.of(
-            "SELECT "+ schema +".create_hypertable( ? , ? , chunk_time_interval => INTERVAL '" + interval + "')",
+            "SELECT " + functionSchema + ".create_hypertable( ? , ? , chunk_time_interval => INTERVAL '" + interval + "')",
             table.getFullName(),
             table.getColumnNow(createHypertable.getColumn()).getName()
         );

jetlinks-components/timescaledb-component/src/main/java/org/jetlinks/community/timescaledb/metadata/TimescaleDBDialectProvider.java

@@ -49,7 +49,7 @@ public String getBindSymbol() {
     @Override
     public RDBSchemaMetadata createSchema(String name) {
         PostgresqlSchemaMetadata schema = new PostgresqlSchemaMetadata(name);
-        schema.addFeature(new TimescaleDBCreateTableSqlBuilder(name));
+        schema.addFeature(new TimescaleDBCreateTableSqlBuilder());
         schema.addFeature(new TimescaleDBAlterTableSqlBuilder());
         DefaultValueCodecFactory codecFactory = new DefaultValueCodecFactory();
         codecFactory

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

@@ -28,6 +28,7 @@
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.hswebframework.web.api.crud.entity.QueryParamEntity;
 import org.hswebframework.web.crud.query.QueryHelper;
+import org.jetlinks.community.timescaledb.metadata.FunctionSchema;
 import org.jetlinks.core.metadata.EventMetadata;
 import org.jetlinks.core.things.ThingsRegistry;
 import org.jetlinks.community.things.data.AggregationRequest;
@@ -122,7 +123,8 @@ static Flux<AggregationData> doAggregation0(DatabaseOperator database,
         if (request.getInterval() != null) {
             NativeSelectColumn column = createTimeGroupColumn(
                 request.getFrom().getTime(),
-                request.getInterval()
+                request.getInterval(),
+                database.getMetadata().getTable(metric).get().getFeatureNow(FunctionSchema.ID).getFunctionSchema()
             );
             query.groupBy(column);
             query.select(column);

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

@@ -28,6 +28,7 @@
 import org.hswebframework.web.api.crud.entity.PagerResult;
 import org.hswebframework.web.api.crud.entity.QueryParamEntity;
 import org.hswebframework.web.crud.query.QueryHelper;
+import org.jetlinks.community.timescaledb.metadata.FunctionSchema;
 import org.jetlinks.core.metadata.EventMetadata;
 import org.jetlinks.core.things.ThingsRegistry;
 import org.jetlinks.community.things.data.AggregationRequest;
@@ -117,7 +118,8 @@ protected Flux<AggregationData> doAggregation(String metric,
         if (request.getInterval() != null) {
             NativeSelectColumn column = TimescaleDBUtils.createTimeGroupColumn(
                 request.getFrom().getTime(),
-                request.getInterval()
+                request.getInterval(),
+                database.getMetadata().getTable(metric).get().getFeatureNow(FunctionSchema.ID).getFunctionSchema()
             );
 
             query.groupBy(column);

jetlinks-components/timescaledb-component/src/main/java/org/jetlinks/community/timescaledb/timeseries/TimescaleDBTimeSeriesManager.java

@@ -20,6 +20,8 @@
 import org.hswebframework.ezorm.rdb.codec.DateTimeCodec;
 import org.hswebframework.ezorm.rdb.metadata.RDBIndexMetadata;
 import org.hswebframework.ezorm.rdb.operator.ddl.TableBuilder;
+import org.jetlinks.community.timescaledb.TimescaleDBProperties;
+import org.jetlinks.community.timescaledb.metadata.FunctionSchema;
 import org.jetlinks.core.metadata.PropertyMetadata;
 import org.jetlinks.community.Interval;
 import org.jetlinks.community.things.data.ThingsDataConstants;
@@ -48,6 +50,8 @@ public class TimescaleDBTimeSeriesManager implements TimeSeriesManager {
 
     private final TimescaleDBOperations operations;
 
+    private final TimescaleDBProperties timescaleDBProperties;
+
     @Override
     public TimeSeriesService getService(TimeSeriesMetric metric) {
         return getService(metric.getId());
@@ -82,6 +86,7 @@ public Mono<Void> registerMetadata(TimeSeriesMetadata metadata) {
             .ddl()
             .createOrAlter(tableName)
             .custom(table -> {
+                table.addFeature(FunctionSchema.of(timescaleDBProperties.getFunctionSchema()));
                 table.addFeature(new CreateHypertable(ThingsDataConstants.COLUMN_TIMESTAMP, properties.getChunkTimeInterval()));
                 Interval interval = properties.getRetentionPolicy(tableName);
                 if (interval != null && interval.getNumber().longValue() > 0) {

jetlinks-components/timescaledb-component/src/main/java/org/jetlinks/community/timescaledb/timeseries/TimescaleDBTimeSeriesService.java

@@ -24,6 +24,7 @@
 import org.hswebframework.ezorm.rdb.operator.dml.query.SelectColumn;
 import org.hswebframework.web.bean.FastBeanCopier;
 import org.hswebframework.web.id.IDGenerator;
+import org.jetlinks.community.timescaledb.metadata.FunctionSchema;
 import org.jetlinks.core.utils.Reactors;
 import org.jetlinks.community.things.data.ThingsDataConstants;
 import org.jetlinks.community.timescaledb.TimescaleDBOperations;
@@ -113,7 +114,7 @@ public Flux<AggregationData> aggregation(AggregationQueryParam param) {
         for (Group group : groups) {
             if (group instanceof TimeGroup) {
                 _timeGroup = ((TimeGroup) group);
-                NativeSelectColumn column = TimescaleDBUtils.createTimeGroupColumn(startWith, _timeGroup.getInterval());
+                NativeSelectColumn column = TimescaleDBUtils.createTimeGroupColumn(startWith, _timeGroup.getInterval(),operations.database().getMetadata().getTable(metric).get().getFeatureNow(FunctionSchema.ID).getFunctionSchema());
                 column.setColumn(ThingsDataConstants.COLUMN_TIMESTAMP);
                 column.setAlias(group.getAlias());
                 query.select(column);

jetlinks-standalone/src/main/resources/application-default.yml

@@ -16,6 +16,7 @@ REDIS_DATABASE: 0 # redis 数据库索引
 
 # timescalbedb相关配置
 TIMESCALEDB_SCHEMA: public # timescaledb 数据库schema,默认public
+TIMESCALEDB_FUNCTION_SCHEMA: public # 时序数据所用到相关函数所在的schema
 
 
 # elasticsearch相关配置

jetlinks-standalone/src/main/resources/application.yml

@@ -61,6 +61,7 @@ timescaledb:
     #        username: postgres
     #        password: p@ssw0rd
     schema: ${TIMESCALEDB_SCHEMA:${easyorm.default-schema}} # timescaledb的schema,默认public
+    function-schema: ${TIMESCALEDB_FUNCTION_SCHEMA:${timescaledb.schema}} #时序数据所用到相关函数所在的schema
     time-series:
         enabled: true
         retention-policies: