优化es查询

jetlinks-components/common-component/src/main/java/org/jetlinks/community/utils/ConverterUtils.java

@@ -0,0 +1,178 @@
+package org.jetlinks.community.utils;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.collections4.ComparatorUtils;
+import org.apache.commons.collections4.MapUtils;
+import org.hswebframework.ezorm.core.param.Sort;
+import org.hswebframework.ezorm.core.param.Term;
+import org.hswebframework.web.api.crud.entity.TermExpressionParser;
+import org.hswebframework.web.bean.FastBeanCopier;
+import org.jetlinks.reactor.ql.utils.CompareUtils;
+import reactor.core.publisher.Flux;
+
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class ConverterUtils {
+
+    /**
+     * 尝试转换值为集合,如果不是集合格式则直接返回该值
+     *
+     * @param value     值
+     * @param converter 转换器,用户转换单个结果
+     * @return 转换结果
+     */
+    public static Object tryConvertToList(Object value, Function<Object, Object> converter) {
+        List<Object> list = convertToList(value, converter);
+        if (list.size() == 1) {
+            return list.get(0);
+        }
+        return list;
+    }
+
+    /**
+     * 转换参数为指定类型的List
+     *
+     * @param value     参数
+     * @param converter 类型转换器
+     * @param <T>       List中元素类型
+     * @return 转换后的List
+     */
+    public static <T> List<T> convertToList(Object value, Function<Object, T> converter) {
+        if (value == null) {
+            return Collections.emptyList();
+        }
+        if (value instanceof String) {
+            value = ((String) value).split(",");
+        }
+
+        if (value instanceof Object[]) {
+            value = Arrays.asList(((Object[]) value));
+        }
+        if (value instanceof Collection) {
+            return ((Collection<?>) value)
+                .stream()
+                .map(converter)
+                .collect(Collectors.toList());
+        }
+        return Collections.singletonList(converter.apply(value));
+    }
+
+    /**
+     * 转换参数为List
+     *
+     * @param value 参数
+     * @return 排序后的流
+     */
+    public static List<Object> convertToList(Object value) {
+        return convertToList(value, Function.identity());
+    }
+
+    /**
+     * 根据排序参数对指定对Flux流进行排序
+     *
+     * @param flux  Flux
+     * @param sorts 排序参数
+     * @param <T>   流中元素类型
+     * @return 排序后的流
+     */
+    public static <T> Flux<T> convertSortedStream(Flux<T> flux, Sort... sorts) {
+        return convertSortedStream(flux, Arrays.asList(sorts));
+    }
+
+    /**
+     * 根据排序参数对指定对Flux流进行排序
+     *
+     * @param flux  Flux
+     * @param sorts 排序参数
+     * @param <T>   流中元素类型
+     * @return 排序后的流
+     */
+    public static <T> Flux<T> convertSortedStream(Flux<T> flux, Collection<Sort> sorts) {
+        if (CollectionUtils.isEmpty(sorts)) {
+            return flux;
+        }
+        List<Comparator<T>> comparators = new ArrayList<>(sorts.size());
+        for (Sort sort : sorts) {
+            String column = sort.getName();
+            Comparator<T> comparator = (left, right) -> {
+                Object leftVal = FastBeanCopier.copy(left, new HashMap<>()).get(column);
+                Object rightVal = FastBeanCopier.copy(right, new HashMap<>()).get(column);
+                return CompareUtils.compare(leftVal, rightVal);
+            };
+            if (sort.getOrder().equalsIgnoreCase("desc")) {
+                comparator = comparator.reversed();
+            }
+            comparators.add(comparator);
+
+        }
+        return flux.sort(ComparatorUtils.chainedComparator(comparators));
+    }
+
+    /**
+     * 将Map转为tag,如果map中到值不是数字,则转为json.
+     * <pre>
+     *      {"key1":"value1","key2":["value2"]} => key,value1,key2,["value2"]
+     *  </pre>
+     *
+     * @param map map
+     * @return tags
+     */
+    public static String[] convertMapToTags(Map<String, Object> map) {
+        if (MapUtils.isEmpty(map)) {
+            return new String[0];
+        }
+        String[] tags = new String[map.size() * 2];
+        int index = 0;
+        for (Map.Entry<String, Object> entry : map.entrySet()) {
+            String key = entry.getKey();
+            Object value = entry.getValue();
+            if (value == null) {
+                continue;
+            }
+            String strValue = value instanceof String
+                ? String.valueOf(value)
+                : JSON.toJSONString(value);
+
+            tags[index++] = key;
+            tags[index++] = strValue;
+        }
+        if (tags.length > index) {
+            return Arrays.copyOf(tags, index);
+        }
+        return tags;
+    }
+
+    /**
+     * 将对象转为查询条件,支持json和表达式格式,如:
+     * <pre>
+     *   //name = xxx and age > 10
+     *   convertTerms("name is xxx and age gt 10")
+     *
+     * </pre>
+     *
+     * @param value
+     * @return 条件集合
+     */
+    @SuppressWarnings("all")
+    public static List<Term> convertTerms(Object value) {
+        if (value instanceof String) {
+            String strVal = String.valueOf(value);
+            //json字符串
+            if (strVal.startsWith("[")) {
+                value = JSON.parseArray(strVal);
+            } else {
+                //表达式
+                return TermExpressionParser.parse(strVal);
+            }
+        }
+        if (value instanceof List) {
+            return new JSONArray(((List) value)).toJavaList(Term.class);
+        } else {
+            throw new UnsupportedOperationException("unsupported term value:" + value);
+        }
+    }
+}

jetlinks-components/common-component/src/main/java/org/jetlinks/community/utils/TimeUtils.java

@@ -1,15 +1,13 @@
 package org.jetlinks.community.utils;
 
+import org.jetlinks.reactor.ql.utils.CastUtils;
+import reactor.core.publisher.Flux;
+
 import java.math.BigDecimal;
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.Date;
 
-/**
- * 时间工具类
- *
- * @author zhouhao
- */
 public class TimeUtils {
 
 
@@ -54,17 +52,25 @@ public class TimeUtils {
                 duration = duration.plus(plus);
             }
         }
+        if (numIndex != 0) {
+            duration = duration.plus(Duration.ofMillis(new BigDecimal(tmp, 0, numIndex).longValue()));
+        }
         return duration;
     }
 
+
     public static ChronoUnit parseUnit(String expr) {
 
         expr = expr.toUpperCase();
 
+        if (expr.equals("MILLENNIA")) {
+            return ChronoUnit.MILLENNIA;
+        } else if (expr.equals("FOREVER")) {
+            return ChronoUnit.FOREVER;
+        }
         if (!expr.endsWith("S")) {
             expr = expr + "S";
         }
-
         return ChronoUnit.valueOf(expr);
 
     }
@@ -85,4 +91,43 @@ public class TimeUtils {
         return new Date(DateMathParser.parse(expr, System::currentTimeMillis));
     }
 
+    public static Date convertToDate(Object obj) {
+        if(obj instanceof String){
+            return new Date(DateMathParser.parse(String.valueOf(obj), System::currentTimeMillis));
+        }
+        return CastUtils.castDate(obj);
+    }
+
+
+    public static long round(long ts, long interval) {
+        return (ts / interval) * interval;
+    }
+
+    /**
+     * 解析指定时间区间为每一个间隔时间
+     *
+     * @param from     时间从
+     * @param to       时间到
+     * @param interval 间隔
+     * @return 时间
+     */
+    public static Flux<Long> parseIntervalRange(long from, long to, long interval) {
+        return Flux
+            .create(sink -> {
+                long _from = from, _to = to;
+                if (_from > _to) {
+                    _from = to;
+                    _to = from;
+                }
+                _from = round(_from, interval);
+                _to = round(_to, interval);
+                sink.next(_from);
+                while (_from < _to && !sink.isCancelled()) {
+                    _from = _from + interval;
+                    sink.next(_from);
+                }
+                sink.complete();
+            });
+
+    }
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/parser/DefaultTermTypeParser.java

@@ -1,6 +1,8 @@
 package org.jetlinks.community.elastic.search.parser;
 
+import org.apache.lucene.search.join.ScoreMode;
 import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.NestedQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.hswebframework.ezorm.core.param.Term;
@@ -23,6 +25,15 @@ public class DefaultTermTypeParser implements TermTypeParser {
 
 
     private QueryBuilder queryBuilder(Term term) {
-        return TermTypeEnum.of(term.getTermType().trim()).map(e -> e.process(term)).orElse(QueryBuilders.boolQuery());
+        return TermTypeEnum.of(term.getTermType().trim())
+                           .map(e -> createQueryBuilder(e,term))
+                           .orElse(QueryBuilders.boolQuery());
+    }
+
+    static QueryBuilder createQueryBuilder(TermTypeEnum linkTypeEnum, Term term) {
+        if (term.getColumn().contains(".")) {
+            return new NestedQueryBuilder(term.getColumn().split("[.]")[0], linkTypeEnum.process(term), ScoreMode.Max);
+        }
+        return linkTypeEnum.process(term);
     }
 }

jetlinks-components/elasticsearch-component/src/main/java/org/jetlinks/community/elastic/search/utils/QueryParamTranslator.java

@@ -11,8 +11,13 @@ import org.hswebframework.ezorm.core.param.Sort;
 import org.hswebframework.ezorm.core.param.Term;
 import org.jetlinks.community.elastic.search.index.ElasticSearchIndexMetadata;
 import org.jetlinks.community.elastic.search.parser.DefaultLinkTypeParser;
+import org.jetlinks.community.utils.ConverterUtils;
+import org.jetlinks.community.utils.TimeUtils;
+import org.jetlinks.core.metadata.Converter;
 import org.jetlinks.core.metadata.DataType;
 import org.jetlinks.core.metadata.PropertyMetadata;
+import org.jetlinks.core.metadata.types.DateTimeType;
+import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
 import java.util.Map;
@@ -43,12 +48,21 @@ public class QueryParamTranslator {
         Consumer<Term> paramConverter = doNotingParamConverter;
         if (metadata != null) {
             paramConverter = t -> {
-                if (StringUtils.isEmpty(t.getColumn())) {
+                if (ObjectUtils.isEmpty(t.getColumn())) {
                     return;
                 }
                 PropertyMetadata property = metadata.getProperty(t.getColumn());
                 if (null != property) {
                     DataType type = property.getValueType();
+                    t.setValue(
+                        ConverterUtils.tryConvertToList(t.getValue(), val -> {
+                            if (type instanceof DateTimeType) {
+                                return TimeUtils.convertToDate(val).getTime();
+                            } else if (type instanceof Converter) {
+                                return ((Converter<?>) type).convert(val);
+                            }
+                            return val;
+                        }));
                     converter.getOrDefault(type.getId(), defaultDataTypeConverter).accept(type, t);
                 }
             };
@@ -58,6 +72,7 @@ public class QueryParamTranslator {
         }
         return queryBuilders;
     }
+
     public static SearchSourceBuilder convertSearchSourceBuilder(QueryParam queryParam, ElasticSearchIndexMetadata metadata) {
         SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
         if (queryParam.isPaging()) {