From 57b4b9e3645f22bfeb264b47678810d1e005c37d Mon Sep 17 00:00:00 2001 From: leto-bbq Date: Thu, 27 Aug 2026 18:08:36 +0800 Subject: [PATCH] docs: document IoTDBLocal for table UDFs --- src/.vuepress/sidebar/V2.0.x/en-Table.ts | 2 +- src/.vuepress/sidebar/V2.0.x/zh-Table.ts | 2 +- ...ion.md => User-defined-function_apache.md} | 182 +++++++++++++++++- ...ion.md => User-defined-function_apache.md} | 182 +++++++++++++++++- ...ion.md => User-defined-function_apache.md} | 182 +++++++++++++++++- ...ion.md => User-defined-function_apache.md} | 182 +++++++++++++++++- 6 files changed, 726 insertions(+), 6 deletions(-) rename src/UserGuide/Master/Table/User-Manual/{User-defined-function.md => User-defined-function_apache.md} (77%) rename src/UserGuide/latest-Table/User-Manual/{User-defined-function.md => User-defined-function_apache.md} (77%) rename src/zh/UserGuide/Master/Table/User-Manual/{User-defined-function.md => User-defined-function_apache.md} (77%) rename src/zh/UserGuide/latest-Table/User-Manual/{User-defined-function.md => User-defined-function_apache.md} (77%) diff --git a/src/.vuepress/sidebar/V2.0.x/en-Table.ts b/src/.vuepress/sidebar/V2.0.x/en-Table.ts index 73eee9bcd..830cbe20d 100644 --- a/src/.vuepress/sidebar/V2.0.x/en-Table.ts +++ b/src/.vuepress/sidebar/V2.0.x/en-Table.ts @@ -115,7 +115,7 @@ export const enSidebar = { ], badge: ['hot'], }, - { text: 'UDF', link: 'User-defined-function' }, + { text: 'UDF', link: 'User-defined-function_apache' }, { text: 'Security Management', collapsible: true, diff --git a/src/.vuepress/sidebar/V2.0.x/zh-Table.ts b/src/.vuepress/sidebar/V2.0.x/zh-Table.ts index a1f69708b..3ca76df40 100644 --- a/src/.vuepress/sidebar/V2.0.x/zh-Table.ts +++ b/src/.vuepress/sidebar/V2.0.x/zh-Table.ts @@ -115,7 +115,7 @@ export const zhSidebar = { ], badge: ['hot'], }, - { text: 'UDF', link: 'User-defined-function' }, + { text: 'UDF', link: 'User-defined-function_apache' }, { text: '安全管理', collapsible: true, diff --git a/src/UserGuide/Master/Table/User-Manual/User-defined-function.md b/src/UserGuide/Master/Table/User-Manual/User-defined-function_apache.md similarity index 77% rename from src/UserGuide/Master/Table/User-Manual/User-defined-function.md rename to src/UserGuide/Master/Table/User-Manual/User-defined-function_apache.md index 4213a7f9c..e3fc7383b 100644 --- a/src/UserGuide/Master/Table/User-Manual/User-defined-function.md +++ b/src/UserGuide/Master/Table/User-Manual/User-defined-function_apache.md @@ -448,6 +448,186 @@ The result set of a table function consists of two parts: * If no pass-through but PARTITION BY is specified: Only partition columns are included. * If neither is specified: No additional columns are added. -### 3.5 Complete Maven Project Example +### 3.5 Accessing IoTDB from UDFs + +Starting from **V2.0.11**, table-model UDFs can use `IoTDBLocal` to execute SQL queries and write logs during data processing. `IoTDBLocal` is supplied by the framework. Queries reuse the permissions of the current connection user and return streaming result sets. This capability is included in `udf-api`; no additional dependency is required. + +Existing methods without an `IoTDBLocal` parameter remain available. Implement the overloads with an `IoTDBLocal` parameter only when the UDF needs to query IoTDB or write logs. + +#### 3.5.1 API Reference + +`IoTDBLocal` is in the `org.apache.iotdb.udf.api` package and provides the following capabilities: + +| API | Description | +|---|---| +| `UDFResultSet query(String sql) throws UDFException` | Executes SQL and returns the query result as a streaming result set. The query uses the current connection user's permissions. | +| `info(...)` | Writes an INFO-level log message. Message arguments and exception objects are supported. | +| `warn(...)` | Writes a WARN-level log message. Message arguments and exception objects are supported. | +| `error(...)` | Writes an ERROR-level log message. Message arguments and exception objects are supported. | + +The `UDFResultSet` returned by `query` implements `AutoCloseable` and provides the following methods for reading rows: + +| API | Description | +|---|---| +| `boolean hasNext() throws UDFException` | Checks whether another row is available. | +| `Record next() throws UDFException` | Returns the next row. Throws `NoSuchElementException` when no more rows are available. | +| `void close() throws IoTDBLocalException` | Closes the result set and releases resources. | + +Use try-with-resources to close result sets promptly. Even if a result set is not closed manually, the framework releases any result sets that remain open after the current UDF execution finishes. + +The following table-model UDF methods can receive `IoTDBLocal`: + +| UDF type | Supported methods | +|---|---| +| Scalar function (UDSF) | `beforeStart`, `evaluate`, `beforeDestroy` | +| Aggregate function (UDAF) | `beforeStart`, `addInput`, `combineState`, `outputFinal`, `beforeDestroy` | +| Table function (UDTF) | `beforeStart`, `process`, `finish`, and `beforeDestroy` of `TableFunctionDataProcessor`; `beforeStart`, `process`, and `beforeDestroy` of `TableFunctionLeafProcessor` | + +For example, a scalar function can implement the following overloads as needed: + +```Java +default void beforeStart(FunctionArguments arguments, IoTDBLocal local) + throws UDFException; + +default Object evaluate(Record input, IoTDBLocal local) + throws UDFException; + +default void beforeDestroy(IoTDBLocal local); +``` + +#### 3.5.2 Example + +The following example queries device names and temperature limits in `beforeStart` and caches the results in memory. If the UDF only needs device names, keep the first query. If it needs to combine multiple data sources, call `query` multiple times as shown here. For reference data that does not change during one UDF execution, query it during initialization instead of executing the same SQL for every row in `evaluate`. + +1. Create the sample tables and insert data. + +```SQL +CREATE TABLE readings ( + device_id STRING TAG, + temperature DOUBLE FIELD +); + +CREATE TABLE device_info ( + device_id STRING TAG, + device_name STRING FIELD +); + +CREATE TABLE device_limits ( + device_id STRING TAG, + max_temp DOUBLE FIELD +); + +INSERT INTO device_info(time, device_id, device_name) +VALUES (1, 'd1', 'Workshop 1 temperature sensor'), + (1, 'd2', 'Workshop 2 temperature sensor'); + +INSERT INTO device_limits(time, device_id, max_temp) +VALUES (1, 'd1', 30.0), + (1, 'd2', 35.0); + +INSERT INTO readings(time, device_id, temperature) +VALUES (1000, 'd1', 25.5), + (1001, 'd2', 32.0), + (1002, 'd3', 20.0); +``` + +2. Implement the scalar function. + +```Java +public class DeviceSummaryFunction implements ScalarFunction { + + private Map idToName = Map.of(); + private Map idToMaxTemp = Map.of(); + + @Override + public ScalarFunctionAnalysis analyze(FunctionArguments arguments) + throws UDFArgumentNotValidException { + return new ScalarFunctionAnalysis.Builder() + .outputDataType(Type.STRING) + .build(); + } + + @Override + public void beforeStart(FunctionArguments arguments, IoTDBLocal local) + throws UDFException { + local.info("DeviceSummaryFunction: loading reference data"); + + // Keep only this block when a single query is needed. + Map names = new HashMap<>(); + try (UDFResultSet resultSet = + local.query("SELECT device_id, device_name FROM device_info")) { + while (resultSet.hasNext()) { + Record row = resultSet.next(); + names.put(row.getString(0), row.getString(1)); + } + } + + // Execute additional queries when multiple data sources are needed. + Map limits = new HashMap<>(); + try (UDFResultSet resultSet = + local.query("SELECT device_id, max_temp FROM device_limits")) { + while (resultSet.hasNext()) { + Record row = resultSet.next(); + limits.put(row.getString(0), row.getDouble(1)); + } + } + + idToName = names; + idToMaxTemp = limits; + local.info( + "DeviceSummaryFunction: loaded {} names and {} limits", + idToName.size(), + idToMaxTemp.size()); + } + + @Override + public Object evaluate(Record input) throws UDFException { + return summarize(input); + } + + @Override + public Object evaluate(Record input, IoTDBLocal local) + throws UDFException { + return summarize(input); + } + + private Binary summarize(Record input) { + String deviceId = input.getString(0); + String name = idToName.getOrDefault(deviceId, "Unknown device"); + Double maxTemp = idToMaxTemp.get(deviceId); + + String result = + String.format( + "%s (limit: %s)", + name, + maxTemp == null ? "unknown" : maxTemp); + return new Binary(result, TSFileConfig.STRING_CHARSET); + } +} +``` + +3. Register and use the function. + +```SQL +CREATE FUNCTION device_summary +AS 'org.apache.iotdb.udf.demo.DeviceSummaryFunction'; + +SELECT time, + device_id, + temperature, + device_summary(device_id) AS summary +FROM readings; +``` + +The query returns results similar to the following: + +```SQL +d1 25.5 Workshop 1 temperature sensor (limit: 30.0) +d2 32.0 Workshop 2 temperature sensor (limit: 35.0) +d3 20.0 Unknown device (limit: unknown) +``` + + +### 3.6 Complete Maven Project Example For Maven-based implementations, refer to the sample project: [udf-example](https://github.com/apache/iotdb/tree/master/example/udf). diff --git a/src/UserGuide/latest-Table/User-Manual/User-defined-function.md b/src/UserGuide/latest-Table/User-Manual/User-defined-function_apache.md similarity index 77% rename from src/UserGuide/latest-Table/User-Manual/User-defined-function.md rename to src/UserGuide/latest-Table/User-Manual/User-defined-function_apache.md index 770151fd5..e88a688de 100644 --- a/src/UserGuide/latest-Table/User-Manual/User-defined-function.md +++ b/src/UserGuide/latest-Table/User-Manual/User-defined-function_apache.md @@ -448,6 +448,186 @@ The result set of a table function consists of two parts: * If no pass-through but PARTITION BY is specified: Only partition columns are included. * If neither is specified: No additional columns are added. -### 3.5 Complete Maven Project Example +### 3.5 Accessing IoTDB from UDFs + +Starting from **V2.0.11**, table-model UDFs can use `IoTDBLocal` to execute SQL queries and write logs during data processing. `IoTDBLocal` is supplied by the framework. Queries reuse the permissions of the current connection user and return streaming result sets. This capability is included in `udf-api`; no additional dependency is required. + +Existing methods without an `IoTDBLocal` parameter remain available. Implement the overloads with an `IoTDBLocal` parameter only when the UDF needs to query IoTDB or write logs. + +#### 3.5.1 API Reference + +`IoTDBLocal` is in the `org.apache.iotdb.udf.api` package and provides the following capabilities: + +| API | Description | +|---|---| +| `UDFResultSet query(String sql) throws UDFException` | Executes SQL and returns the query result as a streaming result set. The query uses the current connection user's permissions. | +| `info(...)` | Writes an INFO-level log message. Message arguments and exception objects are supported. | +| `warn(...)` | Writes a WARN-level log message. Message arguments and exception objects are supported. | +| `error(...)` | Writes an ERROR-level log message. Message arguments and exception objects are supported. | + +The `UDFResultSet` returned by `query` implements `AutoCloseable` and provides the following methods for reading rows: + +| API | Description | +|---|---| +| `boolean hasNext() throws UDFException` | Checks whether another row is available. | +| `Record next() throws UDFException` | Returns the next row. Throws `NoSuchElementException` when no more rows are available. | +| `void close() throws IoTDBLocalException` | Closes the result set and releases resources. | + +Use try-with-resources to close result sets promptly. Even if a result set is not closed manually, the framework releases any result sets that remain open after the current UDF execution finishes. + +The following table-model UDF methods can receive `IoTDBLocal`: + +| UDF type | Supported methods | +|---|---| +| Scalar function (UDSF) | `beforeStart`, `evaluate`, `beforeDestroy` | +| Aggregate function (UDAF) | `beforeStart`, `addInput`, `combineState`, `outputFinal`, `beforeDestroy` | +| Table function (UDTF) | `beforeStart`, `process`, `finish`, and `beforeDestroy` of `TableFunctionDataProcessor`; `beforeStart`, `process`, and `beforeDestroy` of `TableFunctionLeafProcessor` | + +For example, a scalar function can implement the following overloads as needed: + +```Java +default void beforeStart(FunctionArguments arguments, IoTDBLocal local) + throws UDFException; + +default Object evaluate(Record input, IoTDBLocal local) + throws UDFException; + +default void beforeDestroy(IoTDBLocal local); +``` + +#### 3.5.2 Example + +The following example queries device names and temperature limits in `beforeStart` and caches the results in memory. If the UDF only needs device names, keep the first query. If it needs to combine multiple data sources, call `query` multiple times as shown here. For reference data that does not change during one UDF execution, query it during initialization instead of executing the same SQL for every row in `evaluate`. + +1. Create the sample tables and insert data. + +```SQL +CREATE TABLE readings ( + device_id STRING TAG, + temperature DOUBLE FIELD +); + +CREATE TABLE device_info ( + device_id STRING TAG, + device_name STRING FIELD +); + +CREATE TABLE device_limits ( + device_id STRING TAG, + max_temp DOUBLE FIELD +); + +INSERT INTO device_info(time, device_id, device_name) +VALUES (1, 'd1', 'Workshop 1 temperature sensor'), + (1, 'd2', 'Workshop 2 temperature sensor'); + +INSERT INTO device_limits(time, device_id, max_temp) +VALUES (1, 'd1', 30.0), + (1, 'd2', 35.0); + +INSERT INTO readings(time, device_id, temperature) +VALUES (1000, 'd1', 25.5), + (1001, 'd2', 32.0), + (1002, 'd3', 20.0); +``` + +2. Implement the scalar function. + +```Java +public class DeviceSummaryFunction implements ScalarFunction { + + private Map idToName = Map.of(); + private Map idToMaxTemp = Map.of(); + + @Override + public ScalarFunctionAnalysis analyze(FunctionArguments arguments) + throws UDFArgumentNotValidException { + return new ScalarFunctionAnalysis.Builder() + .outputDataType(Type.STRING) + .build(); + } + + @Override + public void beforeStart(FunctionArguments arguments, IoTDBLocal local) + throws UDFException { + local.info("DeviceSummaryFunction: loading reference data"); + + // Keep only this block when a single query is needed. + Map names = new HashMap<>(); + try (UDFResultSet resultSet = + local.query("SELECT device_id, device_name FROM device_info")) { + while (resultSet.hasNext()) { + Record row = resultSet.next(); + names.put(row.getString(0), row.getString(1)); + } + } + + // Execute additional queries when multiple data sources are needed. + Map limits = new HashMap<>(); + try (UDFResultSet resultSet = + local.query("SELECT device_id, max_temp FROM device_limits")) { + while (resultSet.hasNext()) { + Record row = resultSet.next(); + limits.put(row.getString(0), row.getDouble(1)); + } + } + + idToName = names; + idToMaxTemp = limits; + local.info( + "DeviceSummaryFunction: loaded {} names and {} limits", + idToName.size(), + idToMaxTemp.size()); + } + + @Override + public Object evaluate(Record input) throws UDFException { + return summarize(input); + } + + @Override + public Object evaluate(Record input, IoTDBLocal local) + throws UDFException { + return summarize(input); + } + + private Binary summarize(Record input) { + String deviceId = input.getString(0); + String name = idToName.getOrDefault(deviceId, "Unknown device"); + Double maxTemp = idToMaxTemp.get(deviceId); + + String result = + String.format( + "%s (limit: %s)", + name, + maxTemp == null ? "unknown" : maxTemp); + return new Binary(result, TSFileConfig.STRING_CHARSET); + } +} +``` + +3. Register and use the function. + +```SQL +CREATE FUNCTION device_summary +AS 'org.apache.iotdb.udf.demo.DeviceSummaryFunction'; + +SELECT time, + device_id, + temperature, + device_summary(device_id) AS summary +FROM readings; +``` + +The query returns results similar to the following: + +```SQL +d1 25.5 Workshop 1 temperature sensor (limit: 30.0) +d2 32.0 Workshop 2 temperature sensor (limit: 35.0) +d3 20.0 Unknown device (limit: unknown) +``` + + +### 3.6 Complete Maven Project Example For Maven-based implementations, refer to the sample project: [udf-example](https://github.com/apache/iotdb/tree/master/example/udf). diff --git a/src/zh/UserGuide/Master/Table/User-Manual/User-defined-function.md b/src/zh/UserGuide/Master/Table/User-Manual/User-defined-function_apache.md similarity index 77% rename from src/zh/UserGuide/Master/Table/User-Manual/User-defined-function.md rename to src/zh/UserGuide/Master/Table/User-Manual/User-defined-function_apache.md index 288071f3f..42fd233b5 100644 --- a/src/zh/UserGuide/Master/Table/User-Manual/User-defined-function.md +++ b/src/zh/UserGuide/Master/Table/User-Manual/User-defined-function_apache.md @@ -432,6 +432,186 @@ IoTDB 中的表函数为多态表值函数，支持参数类型如下所示： -### 3.5 完整Maven项目示例 +### 3.5 在 UDF 中访问 IoTDB + +自 **V2.0.11** 起，表模型 UDF 可以通过 `IoTDBLocal` 在数据处理过程中执行 SQL 查询和记录日志。`IoTDBLocal` 由框架传入，查询会复用当前连接用户的权限，并以流式结果集返回数据。该功能包含在 `udf-api` 中，无需引入其他依赖。 + +原有不带 `IoTDBLocal` 参数的接口仍然可用。仅当 UDF 需要查询 IoTDB 或记录日志时，才需要实现带 `IoTDBLocal` 参数的重载方法。 + +#### 3.5.1 接口说明 + +`IoTDBLocal` 位于 `org.apache.iotdb.udf.api` 包中，提供以下能力： + +|接口|说明| +|---|---| +|`UDFResultSet query(String sql) throws UDFException`|执行一条 SQL，并以流式结果集返回查询结果。查询使用当前连接用户的权限。| +|`info(...)`|记录 INFO 级别日志，支持消息、格式化参数或异常对象。| +|`warn(...)`|记录 WARN 级别日志，支持消息、格式化参数或异常对象。| +|`error(...)`|记录 ERROR 级别日志，支持消息、格式化参数或异常对象。| + +`query` 返回的 `UDFResultSet` 实现了 `AutoCloseable`，通过以下方法逐行读取结果： + +|接口|说明| +|---|---| +|`boolean hasNext() throws UDFException`|判断结果集中是否还有下一行。| +|`Record next() throws UDFException`|获取下一行；没有更多数据时抛出 `NoSuchElementException`。| +|`void close() throws IoTDBLocalException`|关闭结果集并释放资源。| + +建议使用 `try-with-resources` 及时关闭结果集。即使没有手动关闭，框架也会在本次 UDF 执行结束后释放仍未关闭的结果集。 + +不同类型的表模型 UDF 可以在以下方法中获取 `IoTDBLocal`： + +|UDF 类型|支持的方法| +|---|---| +|标量函数（UDSF）|`beforeStart`、`evaluate`、`beforeDestroy`| +|聚合函数（UDAF）|`beforeStart`、`addInput`、`combineState`、`outputFinal`、`beforeDestroy`| +|表函数（UDTF）|`TableFunctionDataProcessor` 的 `beforeStart`、`process`、`finish`、`beforeDestroy`；
`TableFunctionLeafProcessor` 的 `beforeStart`、`process`、`beforeDestroy`| + +例如，标量函数可以按需实现以下重载方法： + +```Java +default void beforeStart(FunctionArguments arguments, IoTDBLocal local) + throws UDFException; + +default Object evaluate(Record input, IoTDBLocal local) + throws UDFException; + +default void beforeDestroy(IoTDBLocal local); +``` + +#### 3.5.2 使用示例 + +以下示例在 `beforeStart` 中查询设备名称和温度上限，并将结果缓存在内存中。如果 UDF 只需要设备名称，保留第一条查询即可；如果需要组合多个数据源，则可以像本例一样多次调用 `query`。对于不会在一次 UDF 执行期间变化的参考数据，建议在初始化阶段集中查询，避免在 `evaluate` 中对每一行重复执行 SQL。 + +1. 创建示例表并写入数据。 + +```SQL +CREATE TABLE readings ( + device_id STRING TAG, + temperature DOUBLE FIELD +); + +CREATE TABLE device_info ( + device_id STRING TAG, + device_name STRING FIELD +); + +CREATE TABLE device_limits ( + device_id STRING TAG, + max_temp DOUBLE FIELD +); + +INSERT INTO device_info(time, device_id, device_name) +VALUES (1, 'd1', '一号车间温度传感器'), + (1, 'd2', '二号车间温度传感器'); + +INSERT INTO device_limits(time, device_id, max_temp) +VALUES (1, 'd1', 30.0), + (1, 'd2', 35.0); + +INSERT INTO readings(time, device_id, temperature) +VALUES (1000, 'd1', 25.5), + (1001, 'd2', 32.0), + (1002, 'd3', 20.0); +``` + +2. 实现标量函数。 + +```Java +public class DeviceSummaryFunction implements ScalarFunction { + + private Map idToName = Map.of(); + private Map idToMaxTemp = Map.of(); + + @Override + public ScalarFunctionAnalysis analyze(FunctionArguments arguments) + throws UDFArgumentNotValidException { + return new ScalarFunctionAnalysis.Builder() + .outputDataType(Type.STRING) + .build(); + } + + @Override + public void beforeStart(FunctionArguments arguments, IoTDBLocal local) + throws UDFException { + local.info("DeviceSummaryFunction: loading reference data"); + + // 单条查询场景只需保留这一段。 + Map names = new HashMap<>(); + try (UDFResultSet resultSet = + local.query("SELECT device_id, device_name FROM device_info")) { + while (resultSet.hasNext()) { + Record row = resultSet.next(); + names.put(row.getString(0), row.getString(1)); + } + } + + // 需要组合多个数据源时，可以继续执行其他查询。 + Map limits = new HashMap<>(); + try (UDFResultSet resultSet = + local.query("SELECT device_id, max_temp FROM device_limits")) { + while (resultSet.hasNext()) { + Record row = resultSet.next(); + limits.put(row.getString(0), row.getDouble(1)); + } + } + + idToName = names; + idToMaxTemp = limits; + local.info( + "DeviceSummaryFunction: loaded {} names and {} limits", + idToName.size(), + idToMaxTemp.size()); + } + + @Override + public Object evaluate(Record input) throws UDFException { + return summarize(input); + } + + @Override + public Object evaluate(Record input, IoTDBLocal local) + throws UDFException { + return summarize(input); + } + + private Binary summarize(Record input) { + String deviceId = input.getString(0); + String name = idToName.getOrDefault(deviceId, "未知设备"); + Double maxTemp = idToMaxTemp.get(deviceId); + + String result = + String.format( + "%s（上限：%s）", + name, + maxTemp == null ? "未知" : maxTemp); + return new Binary(result, TSFileConfig.STRING_CHARSET); + } +} +``` + +3. 注册并使用该函数。 + +```SQL +CREATE FUNCTION device_summary +AS 'org.apache.iotdb.udf.demo.DeviceSummaryFunction'; + +SELECT time, + device_id, + temperature, + device_summary(device_id) AS summary +FROM readings; +``` + +上述查询将会返回类似如下结果： + +```SQL +d1 25.5 一号车间温度传感器（上限：30.0） +d2 32.0 二号车间温度传感器（上限：35.0） +d3 20.0 未知设备（上限：未知） +``` + + +### 3.6 完整Maven项目示例 如果使用 [Maven](http://search.maven.org/)，可以参考示例项目[udf-example](https://github.com/apache/iotdb/tree/master/example/udf)。 diff --git a/src/zh/UserGuide/latest-Table/User-Manual/User-defined-function.md b/src/zh/UserGuide/latest-Table/User-Manual/User-defined-function_apache.md similarity index 77% rename from src/zh/UserGuide/latest-Table/User-Manual/User-defined-function.md rename to src/zh/UserGuide/latest-Table/User-Manual/User-defined-function_apache.md index 288071f3f..42fd233b5 100644 --- a/src/zh/UserGuide/latest-Table/User-Manual/User-defined-function.md +++ b/src/zh/UserGuide/latest-Table/User-Manual/User-defined-function_apache.md @@ -432,6 +432,186 @@ IoTDB 中的表函数为多态表值函数，支持参数类型如下所示： -### 3.5 完整Maven项目示例 +### 3.5 在 UDF 中访问 IoTDB + +自 **V2.0.11** 起，表模型 UDF 可以通过 `IoTDBLocal` 在数据处理过程中执行 SQL 查询和记录日志。`IoTDBLocal` 由框架传入，查询会复用当前连接用户的权限，并以流式结果集返回数据。该功能包含在 `udf-api` 中，无需引入其他依赖。 + +原有不带 `IoTDBLocal` 参数的接口仍然可用。仅当 UDF 需要查询 IoTDB 或记录日志时，才需要实现带 `IoTDBLocal` 参数的重载方法。 + +#### 3.5.1 接口说明 + +`IoTDBLocal` 位于 `org.apache.iotdb.udf.api` 包中，提供以下能力： + +|接口|说明| +|---|---| +|`UDFResultSet query(String sql) throws UDFException`|执行一条 SQL，并以流式结果集返回查询结果。查询使用当前连接用户的权限。| +|`info(...)`|记录 INFO 级别日志，支持消息、格式化参数或异常对象。| +|`warn(...)`|记录 WARN 级别日志，支持消息、格式化参数或异常对象。| +|`error(...)`|记录 ERROR 级别日志，支持消息、格式化参数或异常对象。| + +`query` 返回的 `UDFResultSet` 实现了 `AutoCloseable`，通过以下方法逐行读取结果： + +|接口|说明| +|---|---| +|`boolean hasNext() throws UDFException`|判断结果集中是否还有下一行。| +|`Record next() throws UDFException`|获取下一行；没有更多数据时抛出 `NoSuchElementException`。| +|`void close() throws IoTDBLocalException`|关闭结果集并释放资源。| + +建议使用 `try-with-resources` 及时关闭结果集。即使没有手动关闭，框架也会在本次 UDF 执行结束后释放仍未关闭的结果集。 + +不同类型的表模型 UDF 可以在以下方法中获取 `IoTDBLocal`： + +|UDF 类型|支持的方法| +|---|---| +|标量函数（UDSF）|`beforeStart`、`evaluate`、`beforeDestroy`| +|聚合函数（UDAF）|`beforeStart`、`addInput`、`combineState`、`outputFinal`、`beforeDestroy`| +|表函数（UDTF）|`TableFunctionDataProcessor` 的 `beforeStart`、`process`、`finish`、`beforeDestroy`；
`TableFunctionLeafProcessor` 的 `beforeStart`、`process`、`beforeDestroy`| + +例如，标量函数可以按需实现以下重载方法： + +```Java +default void beforeStart(FunctionArguments arguments, IoTDBLocal local) + throws UDFException; + +default Object evaluate(Record input, IoTDBLocal local) + throws UDFException; + +default void beforeDestroy(IoTDBLocal local); +``` + +#### 3.5.2 使用示例 + +以下示例在 `beforeStart` 中查询设备名称和温度上限，并将结果缓存在内存中。如果 UDF 只需要设备名称，保留第一条查询即可；如果需要组合多个数据源，则可以像本例一样多次调用 `query`。对于不会在一次 UDF 执行期间变化的参考数据，建议在初始化阶段集中查询，避免在 `evaluate` 中对每一行重复执行 SQL。 + +1. 创建示例表并写入数据。 + +```SQL +CREATE TABLE readings ( + device_id STRING TAG, + temperature DOUBLE FIELD +); + +CREATE TABLE device_info ( + device_id STRING TAG, + device_name STRING FIELD +); + +CREATE TABLE device_limits ( + device_id STRING TAG, + max_temp DOUBLE FIELD +); + +INSERT INTO device_info(time, device_id, device_name) +VALUES (1, 'd1', '一号车间温度传感器'), + (1, 'd2', '二号车间温度传感器'); + +INSERT INTO device_limits(time, device_id, max_temp) +VALUES (1, 'd1', 30.0), + (1, 'd2', 35.0); + +INSERT INTO readings(time, device_id, temperature) +VALUES (1000, 'd1', 25.5), + (1001, 'd2', 32.0), + (1002, 'd3', 20.0); +``` + +2. 实现标量函数。 + +```Java +public class DeviceSummaryFunction implements ScalarFunction { + + private Map idToName = Map.of(); + private Map idToMaxTemp = Map.of(); + + @Override + public ScalarFunctionAnalysis analyze(FunctionArguments arguments) + throws UDFArgumentNotValidException { + return new ScalarFunctionAnalysis.Builder() + .outputDataType(Type.STRING) + .build(); + } + + @Override + public void beforeStart(FunctionArguments arguments, IoTDBLocal local) + throws UDFException { + local.info("DeviceSummaryFunction: loading reference data"); + + // 单条查询场景只需保留这一段。 + Map names = new HashMap<>(); + try (UDFResultSet resultSet = + local.query("SELECT device_id, device_name FROM device_info")) { + while (resultSet.hasNext()) { + Record row = resultSet.next(); + names.put(row.getString(0), row.getString(1)); + } + } + + // 需要组合多个数据源时，可以继续执行其他查询。 + Map limits = new HashMap<>(); + try (UDFResultSet resultSet = + local.query("SELECT device_id, max_temp FROM device_limits")) { + while (resultSet.hasNext()) { + Record row = resultSet.next(); + limits.put(row.getString(0), row.getDouble(1)); + } + } + + idToName = names; + idToMaxTemp = limits; + local.info( + "DeviceSummaryFunction: loaded {} names and {} limits", + idToName.size(), + idToMaxTemp.size()); + } + + @Override + public Object evaluate(Record input) throws UDFException { + return summarize(input); + } + + @Override + public Object evaluate(Record input, IoTDBLocal local) + throws UDFException { + return summarize(input); + } + + private Binary summarize(Record input) { + String deviceId = input.getString(0); + String name = idToName.getOrDefault(deviceId, "未知设备"); + Double maxTemp = idToMaxTemp.get(deviceId); + + String result = + String.format( + "%s（上限：%s）", + name, + maxTemp == null ? "未知" : maxTemp); + return new Binary(result, TSFileConfig.STRING_CHARSET); + } +} +``` + +3. 注册并使用该函数。 + +```SQL +CREATE FUNCTION device_summary +AS 'org.apache.iotdb.udf.demo.DeviceSummaryFunction'; + +SELECT time, + device_id, + temperature, + device_summary(device_id) AS summary +FROM readings; +``` + +上述查询将会返回类似如下结果： + +```SQL +d1 25.5 一号车间温度传感器（上限：30.0） +d2 32.0 二号车间温度传感器（上限：35.0） +d3 20.0 未知设备（上限：未知） +``` + + +### 3.6 完整Maven项目示例 如果使用 [Maven](http://search.maven.org/)，可以参考示例项目[udf-example](https://github.com/apache/iotdb/tree/master/example/udf)。