feat(unnest): 增加表函数平铺支持

src/main/java/org/jetlinks/reactor/ql/DefaultReactorQL.java

@@ -218,6 +218,20 @@ else if ((from instanceof Table)) {
                                 .newRecord(alias, right, left.getContext())
                                 .addRecords(left.getRecords(false)));
             }
+            // join unnest(...), explode(...) or other table functions
+            else if (from instanceof TableFunction) {
+                Function<ReactorQLContext, Flux<ReactorQLRecord>> fromMapper =
+                        FromFeature.createFromMapperByFrom(from, metadata);
+                rightStreamGetter = left -> fromMapper
+                        .apply(left
+                                       .getContext()
+                                       .transfer((name, flux) -> flux
+                                               .map(source -> ReactorQLRecord
+                                                       .newRecord(name, source, left.getContext())
+                                                       .addRecords(left.getRecords(false))))
+                                       .bindAll(left.getRecords(true)))
+                        .map(right -> right.addRecords(left.getRecords(false)));
+            }
             if (rightStreamGetter == null) {
                 throw ReactorQLException.unsupportedFrom(from);
             }

src/main/java/org/jetlinks/reactor/ql/feature/ValueMapFeature.java

@@ -40,6 +40,7 @@
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /**
  * 值转换支持,用来创建数据转换函数.
@@ -88,7 +89,9 @@ public void visit(net.sf.jsqlparser.expression.Function function) {
                 if (name != null) {
                     metadata.getFeature(FeatureId.ValueMap.of(name))
                             .ifPresent(feature -> ref.set(feature.createMapper(function, metadata)));
-                    if (ref.get() == null && metadata.getFeature(FeatureId.From.of(name)).isPresent()) {
+                    if (ref.get() == null
+                            && metadata.getFeature(FeatureId.From.of(name)).isPresent()
+                            && !metadata.getFeature(FeatureId.ValueFlatMap.of(name)).isPresent()) {
                         throw ReactorQLException.unsupportedExpression(
                                 function,
                                 "该函数应作为表函数放在 FROM 子句中使用，不能直接作为 select 列表达式。",
@@ -136,6 +139,20 @@ public void visit(ArrayExpression arrayExpression) {
 
             }
 
+            //select ARRAY[1,2,3] val
+            @Override
+            public void visit(ArrayConstructor arrayConstructor) {
+                List<Function<ReactorQLRecord, Publisher<?>>> mappers = arrayConstructor
+                        .getExpressions()
+                        .stream()
+                        .map(expression -> createMapperNow(expression, metadata))
+                        .collect(Collectors.toList());
+                ref.set(record -> Flux
+                        .fromIterable(mappers)
+                        .concatMap(mapper -> mapper.apply(record))
+                        .collect(Collectors.toList()));
+            }
+
             // select ()
             @Override
             public void visit(Parenthesis value) {

src/main/java/org/jetlinks/reactor/ql/supports/DefaultReactorQLMetadata.java

@@ -952,6 +952,10 @@ private static ChronoUnit chronoUnit(Object unit) {
         addGlobal(new CombineSelectFeature());
         //from merge_by_key((select * from a),(select * from b),'timestamp')
         addGlobal(new MergeByKeyFeature());
+        //from unnest(new_array(1,2,3)) as t(value)
+        addGlobal(new UnnestFeature("unnest"));
+        //from explode(new_array(1,2,3)) as t(value)
+        addGlobal(new UnnestFeature("explode"));
         //from (values())
         addGlobal(new FromValuesFeature());
         //select collect_list(value)
@@ -1304,6 +1308,10 @@ private static ChronoUnit chronoUnit(Object unit) {
 
         //select new_array(1,2,3);
         addGlobal(new FunctionMapFeature("new_array", 9999, 1, stream -> stream.collect(Collectors.toList())));
+        addGlobal(new FunctionMapFeature("array", 9999, 0, stream -> stream.collect(Collectors.toList())));
+        addGlobal(new FunctionMapFeature("array_value", 9999, 0, stream -> stream.collect(Collectors.toList())));
+        addGlobal(new FunctionMapFeature("list_value", 9999, 0, stream -> stream.collect(Collectors.toList())));
+        addGlobal(new FunctionMapFeature("list_pack", 9999, 0, stream -> stream.collect(Collectors.toList())));
 
         //select new_map('k1',v1,'k2',v2);
         addGlobal(new FunctionMapFeature("new_map", 9999, 1, stream -> stream
@@ -1442,6 +1450,9 @@ private static ChronoUnit chronoUnit(Object unit) {
 
 
         addGlobal(new ArrayValueFlatMapFeature());
+        addGlobal(new ArrayValueFlatMapFeature("unnest"));
+        addGlobal(new ArrayValueFlatMapFeature("explode"));
+        addGlobal(new ArrayValueFlatMapFeature("each"));
     }
 
     /**

src/main/java/org/jetlinks/reactor/ql/supports/fmap/ArrayValueFlatMapFeature.java

@@ -38,9 +38,19 @@ public class ArrayValueFlatMapFeature implements ValueFlatMapFeature {
 
     static String ID = FeatureId.ValueFlatMap.of("flat_array").getId();
 
+    private final String id;
+
+    public ArrayValueFlatMapFeature() {
+        this("flat_array");
+    }
+
+    public ArrayValueFlatMapFeature(String name) {
+        this.id = FeatureId.ValueFlatMap.of(name).getId();
+    }
+
     @Override
     public String getId() {
-        return ID;
+        return id;
     }
 
     @Override

src/main/java/org/jetlinks/reactor/ql/supports/from/UnnestFeature.java

@@ -0,0 +1,191 @@
+/*
+ * Copyright 2025 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.reactor.ql.supports.from;
+
+import net.sf.jsqlparser.expression.Alias;
+import net.sf.jsqlparser.expression.Expression;
+import net.sf.jsqlparser.expression.Function;
+import net.sf.jsqlparser.statement.select.FromItem;
+import net.sf.jsqlparser.statement.select.TableFunction;
+import org.apache.commons.collections.CollectionUtils;
+import org.jetlinks.reactor.ql.ReactorQLContext;
+import org.jetlinks.reactor.ql.ReactorQLMetadata;
+import org.jetlinks.reactor.ql.ReactorQLRecord;
+import org.jetlinks.reactor.ql.exception.ReactorQLException;
+import org.jetlinks.reactor.ql.feature.FeatureId;
+import org.jetlinks.reactor.ql.feature.FromFeature;
+import org.jetlinks.reactor.ql.feature.ValueMapFeature;
+import org.jetlinks.reactor.ql.utils.CastUtils;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * SQL dialect compatible array/list flattening table function.
+ *
+ * <pre>{@code
+ * select * from unnest(new_array(1, 2, 3)) as t(value)
+ * select t.id, u.value from test t cross join unnest(t.values) as u(value)
+ * }</pre>
+ */
+public class UnnestFeature implements FromFeature {
+
+    private final String name;
+    private final String id;
+
+    public UnnestFeature(String name) {
+        this.name = name;
+        this.id = FeatureId.From.of(name).getId();
+    }
+
+    @Override
+    public java.util.function.Function<ReactorQLContext, Flux<ReactorQLRecord>> createFromMapper(FromItem fromItem,
+                                                                                                 ReactorQLMetadata metadata) {
+        TableFunction table = (TableFunction) fromItem;
+        Function function = table.getFunction();
+        List<Expression> expressions = function.getParameters() == null
+                ? Collections.emptyList()
+                : function.getParameters().getExpressions();
+        if (CollectionUtils.isEmpty(expressions)) {
+            throw ReactorQLException.functionArgumentCount(function, 1, 9999, 0);
+        }
+
+        List<java.util.function.Function<ReactorQLRecord, Publisher<?>>> valueMappers = expressions
+                .stream()
+                .map(expression -> ValueMapFeature.createMapperNow(expression, metadata))
+                .collect(Collectors.toList());
+
+        String alias = table.getAlias() == null ? null : table.getAlias().getName();
+        List<String> columns = getAliasColumns(table.getAlias(), valueMappers.size());
+
+        return ctx -> {
+            ReactorQLRecord baseRecord = ReactorQLRecord
+                    .newRecord(null, null, ctx)
+                    .addRecords(ctx.getParameters());
+            if (valueMappers.size() == 1) {
+                return Flux
+                        .from(valueMappers.get(0).apply(baseRecord))
+                        .as(CastUtils::flatStream)
+                        .map(value -> ReactorQLRecord.newRecord(alias, toSingleRow(columns, value), ctx));
+            }
+            return Flux
+                    .fromIterable(valueMappers)
+                    .concatMap(mapper -> Flux
+                            .from(mapper.apply(baseRecord))
+                            .as(CastUtils::flatStream)
+                            .collectList())
+                    .collectList()
+                    .flatMapMany(values -> toRows(ctx, alias, columns, values));
+        };
+    }
+
+    private Map<String, Object> toSingleRow(List<String> columns, Object value) {
+        Map<String, Object> row = new LinkedHashMap<>();
+        if (value instanceof Map && columns.size() > 1) {
+            putMapColumns(row, columns, (Map<?, ?>) value);
+            return row;
+        }
+        if (value != null) {
+            row.put(columnName(columns, 0), value);
+        }
+        return row;
+    }
+
+    private List<String> getAliasColumns(Alias alias, int parameterSize) {
+        if (alias != null && alias.getAliasColumns() != null) {
+            return alias
+                    .getAliasColumns()
+                    .stream()
+                    .map(column -> column.name)
+                    .collect(Collectors.toList());
+        }
+        if (parameterSize == 1) {
+            return Collections.singletonList(name);
+        }
+        List<String> columns = new ArrayList<>(parameterSize);
+        for (int i = 0; i < parameterSize; i++) {
+            columns.add(name + "_" + i);
+        }
+        return columns;
+    }
+
+    private Flux<ReactorQLRecord> toRows(ReactorQLContext ctx,
+                                         String alias,
+                                         List<String> columns,
+                                         List<List<Object>> values) {
+        int rowSize = values
+                .stream()
+                .mapToInt(List::size)
+                .max()
+                .orElse(0);
+        if (rowSize == 0) {
+            return Flux.empty();
+        }
+        return Flux
+                .range(0, rowSize)
+                .map(index -> ReactorQLRecord.newRecord(alias, toRow(columns, values, index), ctx));
+    }
+
+    private Map<String, Object> toRow(List<String> columns, List<List<Object>> values, int index) {
+        Map<String, Object> row = new LinkedHashMap<>();
+        for (int i = 0; i < values.size(); i++) {
+            Object value = valueAt(values.get(i), index);
+            if (value != null) {
+                row.put(columnName(columns, i), value);
+            }
+        }
+        return row;
+    }
+
+    private void putMapColumns(Map<String, Object> row, List<String> columns, Map<?, ?> value) {
+        if (columns.size() == 2 && value.containsKey("key") && value.containsKey("value")) {
+            Object key = value.get("key");
+            Object val = value.get("value");
+            if (key != null) {
+                row.put(columns.get(0), key);
+            }
+            if (val != null) {
+                row.put(columns.get(1), val);
+            }
+            return;
+        }
+        for (String column : columns) {
+            Object val = value.get(column);
+            if (val != null) {
+                row.put(column, val);
+            }
+        }
+    }
+
+    private String columnName(List<String> columns, int index) {
+        return index < columns.size() ? columns.get(index) : name + "_" + index;
+    }
+
+    private Object valueAt(List<Object> values, int index) {
+        return index < values.size() ? values.get(index) : null;
+    }
+
+    @Override
+    public String getId() {
+        return id;
+    }
+}

src/main/java/org/jetlinks/reactor/ql/supports/map/PropertyMapFeature.java

@@ -35,22 +35,32 @@ public class PropertyMapFeature implements ValueMapFeature {
     @Override
     public Function<ReactorQLRecord, Publisher<?>> createMapper(Expression expression, ReactorQLMetadata metadata) {
         Column column = ((Column) expression);
-        String[] fullName = column.getFullyQualifiedName().split("[.]", 2);
+        String property = SqlUtils.getCleanStr(column.getFullyQualifiedName());
+        String[] fullName = property.split("[.]", 2);
 
         String name = SqlUtils.getCleanStr(fullName.length == 2 ? fullName[1] : fullName[0]);
         String tableName = fullName.length == 1 ? "this" : SqlUtils.getCleanStr(fullName[0]);
 
         PropertyFeature feature = metadata.getFeatureNow(PropertyFeature.ID);
 
-        return ctx -> getProperty(feature, tableName, name, ctx);
+        return ctx -> getProperty(feature, tableName, name, property, ctx);
     }
 
-    private Mono<Object> getProperty(PropertyFeature feature, String tableName, String name, ReactorQLRecord record) {
-        Object temp = record.getRecord(tableName).orElse(null);
+    private Mono<Object> getProperty(PropertyFeature feature,
+                                     String tableName,
+                                     String name,
+                                     String property,
+                                     ReactorQLRecord record) {
+        Object tableRecord = record.getRecord(tableName).orElse(null);
+        Object temp = null;
 
         //尝试获取表数据
-        if (null != temp) {
-            temp = feature.getProperty(name, temp).orElse(null);
+        if (null != tableRecord) {
+            temp = feature.getProperty(name, tableRecord).orElse(null);
+        }
+        if (null == temp && tableRecord == null && property.contains(".")) {
+            // 如果首段没有命中表别名，则按当前行的嵌套属性解析，兼容 payload.value 这类单源写法。
+            temp = feature.getProperty(property, record.getRecord()).orElse(null);
         }
         if (null == temp) {
             temp = feature.getProperty(name, record.asMap()).orElse(null);