feat: 增强函数支持与异常诊断

- 补充 JSON、时间、字符串等常用函数与 select 列解析场景
- 引入结构化 ReactorQLException 和中英文资源，统一用户可见错误建议
- 增强 ORDER BY、merge_by_key、窗口、聚合、JSON 等异常与安全限制测试

Author: zhou-hao

docs/plans/json-functions-support.md

@@ -9,7 +9,8 @@
 ## 目标
 
 - 增加常用数值函数：`round`、`floor`、`ceil`、`abs`、`sqrt`、`pow`、`power`。
-- 增加常用字符串和正则函数：`lower`、`upper`、`length`、`char_length`、`trim`、`ltrim`、`rtrim`、`replace`、`substring`、`regexp_replace`、`regexp_like`、`regexp_extract`、`regexp_substr`。
+- 增加常用字符串和正则函数：`lower`、`upper`、`length`、`char_length`、`trim`、`ltrim`、`rtrim`、`replace`、`substring`、`concat_ws`、`contains`、`instr`、`locate`、`lpad`、`rpad`、`regexp_replace`、`regexp_like`、`regexp_extract`、`regexp_substr`。
+- 增加常用时间序列处理函数：`date_trunc`、`time_bucket`、`to_unixtime`、`to_millis` / `epoch_ms`，并补充 `date_part` / `date_add` / `date_diff` 的 `quarter`、`week`、毫秒等常见单位。
 - 增加 JSONPath 提取函数：`json_get`、`json_path`、`json_extract`、`json_value`、`json_query`、`json_exists`。
 - 覆盖 MySQL 常见 JSON 函数名：`json_unquote`、`json_type`、`json_valid`、`json_length`、`json_keys`、`json_contains`、`json_contains_path`、`json_overlaps`、`json_array`、`json_object`、`json_merge`、`json_merge_preserve`、`json_merge_patch`。
 - 覆盖 PostgreSQL 常见 JSON 函数名：`json_extract_path`、`json_extract_path_text`、`jsonb_extract_path`、`jsonb_extract_path_text`、`json_array_length`、`jsonb_array_length`、`json_object_keys`、`jsonb_object_keys`、`json_typeof`、`to_json`。
@@ -60,8 +61,9 @@
   - 拒绝典型嵌套量词，例如 `(a+)+`、`(.*){...}` 这类容易导致灾难性回溯的表达式。
   - 保留 Java 正则的常用能力，包括捕获组、大小写标志和 `$1` 替换。
 - 字符串生成函数：
-  - `repeat`、`replace`、`regexp_replace` 的结果默认长度限制为 1MB；setting key：`function.maxGeneratedStringLength`。
+  - `repeat`、`replace`、`regexp_replace`、`concat_ws`、`lpad`、`rpad` 的结果默认长度限制为 1MB；setting key：`function.maxGeneratedStringLength`。
   - `split_part` 不再通过 `String#split` 构造完整数组，按目标下标从前或从后扫描，避免大字符串被分隔成大量中间对象。
+  - `lpad` / `rpad` 对空 padding 返回空值，避免为补齐目标长度进入无意义循环；目标长度仍受硬上限保护。
 - 错误语义：
   - 明显危险或资源超限的参数抛出 `UnsupportedOperationException`，让调用方能定位 SQL 参数问题。
   - 常规业务边界，例如 `regexp_extract` 捕获组不存在，返回空值，避免因下标误用中断整条查询。
@@ -92,6 +94,12 @@
 - `json_keys(json[, path])` / `json_object_keys(json)`：对象返回 key 数组，非对象返回空数组。
 - `json_array(...)`：返回参数数组。
 - `json_object(k1, v1, k2, v2, ...)`：返回对象；奇数参数时忽略最后一个孤立 key。
+- `concat_ws(separator, value...)`：按 MySQL / PostgreSQL / Spark 常见语义拼接，跳过空值；`separator` 为空时返回空值。
+- `contains(string, substring)` / `instr(string, substring)` / `locate(substring, string[, position])`：补充 Trino、MySQL、Spark 常用字符串检索函数，位置返回 1-based，未命中返回 0。
+- `lpad(string, length, pad)` / `rpad(string, length, pad)`：按常见 SQL 行为补齐或截断到目标长度；目标长度小于等于 0 时返回空字符串，`pad` 为空时返回空值。
+- `date_trunc(unit, timestamp)`：按 PostgreSQL / Trino / DuckDB 常见语义截断到 `year`、`quarter`、`month`、`week`、`day`、`hour`、`minute`、`second`、`millisecond`、`microsecond`；`week` 使用 ISO 周一作为起点。
+- `time_bucket(interval, timestamp)`：按 DuckDB / Timescale 类时间桶语义返回时间桶起点；`interval` 支持毫秒数字、`1m` / `15 minutes` / `PT1M` 等 Duration 写法，必须大于 0。
+- `to_unixtime(timestamp)` 返回 Unix 秒（含毫秒小数），`to_millis(timestamp)` / `epoch_ms(timestamp)` 返回 Unix 毫秒。
 
 ## 数据库同名函数语义对齐
 

docs/plans/reactorql-exception-diagnostics.md

@@ -0,0 +1,101 @@
+# ReactorQL 结构化异常诊断开发计划
+
+## 背景
+
+当前 ReactorQL 大量错误直接抛出 `UnsupportedOperationException`，错误信息通常只有“Unsupported expression”或“参数数量错误”。
+这类自由文本不利于用户定位 SQL 问题，也不利于大模型根据错误自动修正查询。
+
+## 目标
+
+- 保留现有 `UnsupportedOperationException` 兼容性，不破坏外部调用方和既有测试。
+- 为 SQL 解析、编译、函数参数、资源限制、安全限制等错误提供结构化诊断信息。
+- 尽可能提供 SQL 表达式、行号、列号、原因、建议和示例。
+- 提供国际化扩展点：错误码、默认文案参数，以及默认中英文资源。
+- `suggestion` 和 `example` 只面向 SQL 使用者提供修复建议或支持用法，不暴露类名、注册表、解析树、执行器、缓存等实现细节。
+
+## 异常模型
+
+新增 `ReactorQLException extends UnsupportedOperationException`，核心字段：
+
+- `i18nCode`：稳定错误码，例如 `error.reactorql.function_argument_count`。
+- `i18nArgs`：用于平台或调用方国际化渲染的参数。
+- `line` / `column`：SQL 起始位置；能从 JSqlParser AST 或 ParseException 取得时填充。
+- `expression`：出错 SQL 表达式或原始 SQL。
+- `reason`：具体失败原因。
+- `suggestion`：推荐修复方向，只描述支持用法或安全边界。
+- `example`：可复制的正确写法。
+
+`getMessage()` 输出默认英文诊断文本，`getLocalizedMessage()` 通过 `ResourceBundle` 读取 `i18n/reactorql/messages_*.properties`。
+JetLinks 平台侧也可以直接读取 `getI18nCode()` / `getI18nArgs()`，再走平台 `LocaleUtils` 或统一错误响应体系。
+
+## 首批错误码
+
+- `error.reactorql.syntax`：SQL 解析失败。
+- `error.reactorql.unsupported_expression`：不支持的 select/value 表达式。
+- `error.reactorql.unsupported_condition`：不支持的 where/having 条件。
+- `error.reactorql.unsupported_group_expression`：不支持的 group by 表达式。
+- `error.reactorql.unsupported_from`：不支持的 from 表达式。
+- `error.reactorql.unsupported_flat_map`：不支持的列转行表达式。
+- `error.reactorql.function_argument_count`：函数参数数量错误。
+- `error.reactorql.invalid_argument`：函数参数值、setting 或安全限制错误。
+- `error.reactorql.resource_limit`：输入行数、JSON 文本、输出长度、窗口大小等资源限制错误。
+
+## 位置信息策略
+
+- 编译期表达式：利用 JSqlParser 4.6 的 `ASTNodeAccess#getASTNode()` 和 `SimpleNode#jjtGetFirstToken()` 读取行列。
+- 解析期错误：从 `ParseException.currentToken.next` 读取行列。
+- 当前 `SqlParserUtils.quoteNonAsciiAliases(...)` 可能改写 SQL。第一阶段只保证未改写或位置未受影响的 SQL 能返回准确行列；后续如需精准映射，可为 SQL 改写过程补 offset mapping。
+
+## 落地阶段
+
+### 阶段 1：核心模型和高频入口
+
+已覆盖：
+
+- `DefaultReactorQLMetadata(String sql)`：SQL 解析失败包装成 `ReactorQLException`。
+- `ValueMapFeature#createMapperNow`：不支持的 value/select 表达式。
+- `FilterFeature#createPredicateNow`：不支持的 where/having 条件。
+- `FunctionMapFeature`：函数参数缺失和数量错误。
+- `DefaultReactorQL`：不支持的 select 表达式和 group by 表达式。
+- 常见资源/安全参数错误：正则风险、字符串长度、setting、日期单位、`time_bucket` interval。
+
+### 阶段 2：继续替换重点函数
+
+已覆盖：
+
+- JSONPath 相关安全错误：`JsonFunctionSupport`、`JsonValueSupport`。
+- `DateFormatFeature`、`SingleParameterFunctionMapFeature`、`CoalesceMapFeature`。
+- `OrderBySupport` 的排序窗口、topN、setting 类型和值域限制。
+- `MergeByKeyFeature` 的参数、setting、排序、重复键和单键行数限制。
+- 聚合和列转行的参数错误：`MapAggFeature`、`CollectListAggFeature`、`CollectRowAggMapFeature`、`ArrayValueFlatMapFeature`。
+- FROM 组合函数和子查询集合操作错误：`zip`、`combine`、子查询 set operation。
+- `if`、`_window` 等常用函数的参数数量和值域错误。
+
+文案边界：
+
+- `reason` 可以描述失败原因，但不要求用户理解内部扩展点。
+- `suggestion` 只给出可执行建议，例如“使用 FROM 子句”“增加 LIMIT”“使用简单 JSONPath”“使用 _window('1m')”。
+- `example` 给出 SQL 或 setting 写法，不展示内部类名、方法名、缓存、解析树或执行器。
+
+### 阶段 3：运行时上下文增强
+
+- 对运行期函数求值失败增加 row/parameter 上下文，避免只看到底层 `TypeCastException` 或 `ArithmeticException`。
+- 对响应式链路中传播的异常保留原始 cause 和 SQL 表达式。
+- 评估是否为 `TypeCastException` 增加结构化诊断字段，或在 ValueMap/Filter 边界统一包装。
+
+## 测试要求
+
+- 断言异常仍然是 `UnsupportedOperationException` 的子类。
+- 断言错误码、表达式、原因、建议、示例字段存在。
+- 断言语法错误和表达式错误尽量带行列信息。
+- 断言 `getLocalizedMessage()` 可以读取英文和中文资源。
+- 断言新增错误码在中英文资源中存在。
+- 断言 JSON 非安全路径、ORDER BY 资源限制、日期格式、窗口参数、merge_by_key 参数和运行期排序校验等场景使用结构化异常。
+- 断言对外建议不包含实现细节关键词。
+- 现有错误场景测试保持通过。
+
+当前验证结果：
+
+- `mvn -q test`：260 tests, 0 failures, 0 errors, 0 skipped。
+- JaCoCo：instruction 93.57%，branch 81.43%，line 94.52%，不低于基准分支覆盖率。
+- `git diff --check`：通过。

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

@@ -24,6 +24,7 @@
 import net.sf.jsqlparser.schema.Table;
 import net.sf.jsqlparser.statement.select.*;
 import org.apache.commons.collections.CollectionUtils;
+import org.jetlinks.reactor.ql.exception.ReactorQLException;
 import org.jetlinks.reactor.ql.feature.*;
 import org.jetlinks.reactor.ql.supports.DefaultReactorQLMetadata;
 import org.jetlinks.reactor.ql.utils.CastUtils;
@@ -218,7 +219,7 @@ else if ((from instanceof Table)) {
                                 .addRecords(left.getRecords(false)));
             }
             if (rightStreamGetter == null) {
-                throw new UnsupportedOperationException("不支持的表关联: " + from);
+                throw ReactorQLException.unsupportedFrom(from);
             }
             Function<ReactorQLRecord, Flux<ReactorQLRecord>> fiRightStreamGetter = rightStreamGetter;
             if (joinInfo.isLeft()) {
@@ -328,7 +329,7 @@ else if (groupByExpression instanceof BinaryExpression) {
                                            metadata.getFeatureNow(FeatureId.GroupBy.of(((BinaryExpression) groupByExpression).getStringExpression()),
                                                                   groupByExpression::toString));
                 } else {
-                    throw new UnsupportedOperationException("Unsupported group expression:" + groupByExpression);
+                    throw ReactorQLException.unsupportedGroupExpression(groupByExpression);
                 }
             }
 
@@ -426,7 +427,11 @@ public void visit(SelectExpressionItem item) {
                                        .ifPresent(mapper -> flatMappers.put(fAlias, mapper));
 
                     if (!mappers.containsKey(fAlias) && !aggMapper.containsKey(fAlias) && !flatMappers.containsKey(fAlias)) {
-                        throw new UnsupportedOperationException("Unsupported expression:" + expression);
+                        throw ReactorQLException.unsupportedExpression(
+                                expression,
+                                "select 列必须是普通表达式、聚合函数或当前支持的列转行函数；表函数应放到 FROM 子句中使用。",
+                                "select count(1) total, date_trunc('minute', timestamp) ts from test group by date_trunc('minute', timestamp)"
+                        );
                     }
                 }
 

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

@@ -19,6 +19,7 @@
 import net.sf.jsqlparser.statement.select.Limit;
 import net.sf.jsqlparser.statement.select.OrderByElement;
 import org.apache.commons.collections.CollectionUtils;
+import org.jetlinks.reactor.ql.exception.ReactorQLException;
 import org.jetlinks.reactor.ql.feature.ValueMapFeature;
 import org.jetlinks.reactor.ql.utils.CastUtils;
 import org.jetlinks.reactor.ql.utils.CompareUtils;
@@ -229,8 +230,11 @@ private static Flux<ReactorQLRecord> limitSortRows(Flux<ReactorQLRecord> flux, l
                 .index()
                 .handle((Tuple2<Long, ReactorQLRecord> tuple, SynchronousSink<ReactorQLRecord> sink) -> {
                     if (tuple.getT1() >= maxRows) {
-                        sink.error(new UnsupportedOperationException("order by rows exceeded setting["
-                                                                              + DefaultReactorQL.SETTING_ORDER_BY_MAX_ROWS + "]:" + maxRows));
+                        sink.error(ReactorQLException.resourceLimit(
+                                "ORDER BY 输入行数超过 setting[" + DefaultReactorQL.SETTING_ORDER_BY_MAX_ROWS + "]: " + maxRows,
+                                "为 ORDER BY 增加 LIMIT 以启用 Top-N 排序，或在可信场景下调大 orderBy.maxRows；无限流建议使用窗口排序。",
+                                "select * from test order by timestamp desc limit 100"
+                        ));
                     } else {
                         sink.next(tuple.getT2());
                     }
@@ -242,11 +246,19 @@ private static Long topNSize(Long rowCount, Long offset) {
             return null;
         }
         if (rowCount < 0) {
-            throw new UnsupportedOperationException("invalid limit:" + rowCount);
+            throw ReactorQLException.invalidArgument(
+                    "LIMIT 不能为负数: " + rowCount,
+                    "使用非负整数 LIMIT；如果不需要返回结果，可使用 limit 0。",
+                    "select * from test order by timestamp desc limit 100"
+            );
         }
         long skip = offset == null ? 0 : offset;
         if (skip < 0) {
-            throw new UnsupportedOperationException("invalid offset:" + skip);
+            throw ReactorQLException.invalidArgument(
+                    "LIMIT offset 不能为负数: " + skip,
+                    "使用非负整数 offset，或省略 offset。",
+                    "select * from test order by timestamp desc limit 10, 100"
+            );
         }
         if (Long.MAX_VALUE - skip < rowCount) {
             return Long.MAX_VALUE;
@@ -270,8 +282,11 @@ private static Long limitValue(Limit limit, ReactorQLContext context, boolean ro
 
     private static void assertSortRows(long rows, long maxRows) {
         if (rows > maxRows || rows > Integer.MAX_VALUE) {
-            throw new UnsupportedOperationException("order by buffer size exceeded setting["
-                                                            + DefaultReactorQL.SETTING_ORDER_BY_MAX_ROWS + "]:" + maxRows + ", actual:" + rows);
+            throw ReactorQLException.resourceLimit(
+                    "ORDER BY 缓冲行数超过 setting[" + DefaultReactorQL.SETTING_ORDER_BY_MAX_ROWS + "]: " + maxRows + ", actual=" + rows,
+                    "为 ORDER BY 增加 LIMIT，或在可信场景下调大 orderBy.maxRows；如果只需要局部有序，可设置 orderBy.windowSize。",
+                    "select * from test order by timestamp desc limit 100"
+            );
         }
     }
 
@@ -280,13 +295,27 @@ private static long longSetting(ReactorQLMetadata metadata,
                                     long defaultValue,
                                     long hardMax,
                                     boolean allowZero) {
-        long value = metadata
-                .getSetting(key)
-                .map(CastUtils::castNumber)
-                .map(Number::longValue)
-                .orElse(defaultValue);
+        long value;
+        try {
+            value = metadata
+                    .getSetting(key)
+                    .map(CastUtils::castNumber)
+                    .map(Number::longValue)
+                    .orElse(defaultValue);
+        } catch (RuntimeException e) {
+            throw ReactorQLException.builder(ReactorQLException.INVALID_ARGUMENT)
+                    .reason("ORDER BY setting[" + key + "] 必须是数字")
+                    .suggestion("确认 setting 在允许范围内；orderBy.maxRows 必须为正数，orderBy.windowSize 可为 0 或正数。")
+                    .example("orderBy.maxRows=10000 或 orderBy.windowSize=1000")
+                    .cause(e)
+                    .build();
+        }
         if ((allowZero ? value < 0 : value <= 0) || value > hardMax) {
-            throw new UnsupportedOperationException("invalid setting[" + key + "]:" + value);
+            throw ReactorQLException.invalidArgument(
+                    "非法 ORDER BY setting[" + key + "]: " + value + ", hardMax=" + hardMax,
+                    "确认 setting 在允许范围内；orderBy.maxRows 必须为正数，orderBy.windowSize 可为 0 或正数。",
+                    "orderBy.maxRows=10000 或 orderBy.windowSize=1000"
+            );
         }
         return value;
     }

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

@@ -16,6 +16,7 @@
 package org.jetlinks.reactor.ql;
 
 import net.sf.jsqlparser.statement.select.PlainSelect;
+import org.jetlinks.reactor.ql.exception.ReactorQLException;
 import org.jetlinks.reactor.ql.feature.Feature;
 import org.jetlinks.reactor.ql.feature.FeatureId;
 import org.jetlinks.reactor.ql.utils.CastUtils;
@@ -141,7 +142,11 @@ default <T extends Feature> T getFeatureNow(FeatureId<T> featureId) {
      */
     default <T extends Feature> T getFeatureNow(FeatureId<T> featureId, Supplier<String> errorMessage) {
         return getFeature(featureId)
-                .orElseThrow(() -> new UnsupportedOperationException("unsupported feature: " + errorMessage.get()));
+                .orElseThrow(() -> ReactorQLException.builder(ReactorQLException.UNSUPPORTED_EXPRESSION)
+                        .reason("当前函数、操作符或查询能力不可用")
+                        .suggestion("确认 SQL 使用的是当前运行环境支持的函数、操作符和 FROM 写法；内置函数名大小写不敏感。")
+                        .example(errorMessage.get())
+                        .build());
     }
 
     Collection<Feature> getFeatures();