You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ch...@apache.org on 2024/11/20 03:11:11 UTC
(iotdb) branch udsf created (now c215e63e0be)
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a change to branch udsf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at c215e63e0be fix it
This branch includes the following new commits:
new 90d34e1136c save
new 6a68d33f448 Merge branch 'table_udsf' into udsf
new 499ae65fc08 save
new b8a4fb6ab45 save
new 73215358b7e save
new c215e63e0be fix it
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 02/06: Merge branch 'table_udsf' into udsf
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a commit to branch udsf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 6a68d33f448d96d7be1b0949464483ecb0768d1d
Merge: 90d34e1136c 93ad64b3f1b
Author: Chen YZ <ch...@gmail.com>
AuthorDate: Tue Nov 19 15:45:15 2024 +0800
Merge branch 'table_udsf' into udsf
.../apache/iotdb/udf/ScalarFunctionExample.java | 42 ---
.../confignode/it/IoTDBConfigNodeSnapshotIT.java | 5 +-
.../iotdb/udf/api/access/ColumnToRowIterator.java | 64 -----
.../customizer/config/ScalarFunctionConfig.java | 30 ---
.../customizer/parameter/FunctionParameters.java | 56 ----
.../udf/api/relational/AggregationFunction.java | 19 ++
.../iotdb/udf/api/relational/SQLFunction.java | 19 ++
.../iotdb/udf/api/relational/ScalarFunction.java | 51 ++--
.../iotdb/udf/api/relational/TableFunction.java | 19 ++
.../iotdb/udf/api/relational/access/Record.java | 129 +++++++++
.../org/apache/iotdb/db/exp/ExpScalarFunction.java | 288 ---------------------
.../UserDefineScalarFunctionTransformer.java | 125 ---------
.../java/org/apache/iotdb/commons/udf/UDFType.java | 19 ++
13 files changed, 229 insertions(+), 637 deletions(-)
(iotdb) 05/06: save
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a commit to branch udsf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 73215358b7eb81df29eca570cb4c84c712e994fc
Author: Chen YZ <ch...@gmail.com>
AuthorDate: Tue Nov 19 19:14:52 2024 +0800
save
---
.../plan/relational/sql/parser/AstBuilder.java | 18 +++++++++++++++---
1 file changed, 15 insertions(+), 3 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
index 85d5638a0b0..c079a00439a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
@@ -45,6 +45,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ColumnDefinition;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ComparisonExpression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CountDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateDB;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateFunction;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateIndex;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreatePipe;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreatePipePlugin;
@@ -60,6 +61,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescribeTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DoubleLiteral;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropColumn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropDB;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropFunction;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropIndex;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropPipe;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropPipePlugin;
@@ -130,6 +132,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentUser;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDataNodes;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDevice;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowFunctions;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowIndex;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowPipePlugins;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowPipes;
@@ -163,6 +166,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.util.AstUtil;
import org.apache.iotdb.db.queryengine.plan.statement.StatementType;
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement;
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateFunctionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.sys.FlushStatement;
import org.apache.iotdb.db.queryengine.plan.statement.sys.SetConfigurationStatement;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.TableBuiltinScalarFunction;
@@ -577,7 +581,14 @@ public class AstBuilder extends RelationalSqlBaseVisitor<Node> {
@Override
public Node visitCreateFunctionStatement(RelationalSqlParser.CreateFunctionStatementContext ctx) {
- return super.visitCreateFunctionStatement(ctx);
+ final String udfName = ((Identifier) visit(ctx.udfName)).getValue();
+ final String className = ((Identifier) visit(ctx.className)).getValue();
+ if (ctx.uriClause() == null) {
+ return new CreateFunction(getLocation(ctx),udfName, className);
+ } else {
+ String uriString = parseAndValidateURI(ctx.uriClause());
+ return new CreateFunction(getLocation(ctx),udfName, className, uriString);
+ }
}
@Override
@@ -587,12 +598,13 @@ public class AstBuilder extends RelationalSqlBaseVisitor<Node> {
@Override
public Node visitDropFunctionStatement(RelationalSqlParser.DropFunctionStatementContext ctx) {
- return super.visitDropFunctionStatement(ctx);
+ final String udfName = ((Identifier) visit(ctx.udfName)).getValue();
+ return new DropFunction(getLocation(ctx), udfName);
}
@Override
public Node visitShowFunctionsStatement(RelationalSqlParser.ShowFunctionsStatementContext ctx) {
- return super.visitShowFunctionsStatement(ctx);
+ return new ShowFunctions();
}
@Override
(iotdb) 01/06: save
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a commit to branch udsf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 90d34e1136c9703b2d8e7e73c83aaa62e163d4c7
Author: Chen YZ <ch...@gmail.com>
AuthorDate: Tue Nov 19 15:05:46 2024 +0800
save
---
.../relational/ColumnTransformerBuilder.java | 10 +++++++
.../iotdb/db/queryengine/plan/Coordinator.java | 8 +++++-
.../execution/config/TableConfigTaskVisitor.java | 25 +++++++++++++++++
.../relational/metadata/TableMetadataImpl.java | 6 +++++
.../commons/udf/utils/UDFDataTypeTransformer.java | 31 ++++++++++++++++++++++
5 files changed, 79 insertions(+), 1 deletion(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java
index f2df4aab0ff..26b30bec4f8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.queryengine.execution.relational;
+import org.apache.iotdb.commons.udf.service.UDFManagementService;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
@@ -152,6 +153,8 @@ import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.Tr
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.TrimColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.TryCastFunctionColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.UpperColumnTransformer;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.UserDefineScalarFunctionTransformer;
+import org.apache.iotdb.udf.api.relational.ScalarFunction;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.common.regexp.LikePattern;
@@ -996,6 +999,13 @@ public class ColumnTransformerBuilder
source,
((LongLiteral) children.get(3)).getParsedValue(),
context.sessionInfo.getZoneId());
+ } else if (UDFManagementService.getInstance()
+ .isAssignableFrom(functionName, ScalarFunction.class)) {
+ List<ColumnTransformer> childrenColumnTransformer =
+ children.stream().map(child -> process(child, context)).collect(Collectors.toList());
+ // TODO(UDSF): check the return type of the function
+ return new UserDefineScalarFunctionTransformer(
+ INT32, functionName, children, childrenColumnTransformer);
}
throw new IllegalArgumentException(String.format("Unknown function: %s", functionName));
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
index 24063d89dc3..39c3cf6587c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
@@ -50,9 +50,11 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.TableModelPlanner
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AddColumn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ClearCache;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateDB;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateFunction;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescribeTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropDB;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropFunction;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Flush;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.PipeStatement;
@@ -68,6 +70,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentTimest
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentUser;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDataNodes;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowFunctions;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowRegions;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowTables;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowVariables;
@@ -350,7 +353,10 @@ public class Coordinator {
|| statement instanceof ShowVersion
|| statement instanceof ShowVariables
|| statement instanceof ShowClusterId
- || statement instanceof ShowCurrentTimestamp) {
+ || statement instanceof ShowCurrentTimestamp
+ || statement instanceof CreateFunction
+ || statement instanceof DropFunction
+ || statement instanceof ShowFunctions) {
return new ConfigExecution(
queryContext,
null,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
index ea06a8bc793..cc514c7c04f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.queryengine.plan.execution.config;
+import org.apache.iotdb.common.rpc.thrift.Model;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.commons.schema.table.TsTable;
@@ -29,10 +30,13 @@ import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.protocol.session.IClientSession;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.CreateFunctionTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.CreatePipePluginTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.DropFunctionTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.DropPipePluginTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowClusterIdTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowClusterTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowFunctionsTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowPipePluginsTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowRegionTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowVariablesTask;
@@ -75,6 +79,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AstVisitor;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ClearCache;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ColumnDefinition;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateDB;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateFunction;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreatePipe;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreatePipePlugin;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateTable;
@@ -82,6 +87,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DataType;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DescribeTable;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropColumn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropDB;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropFunction;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropPipe;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropPipePlugin;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DropTable;
@@ -106,6 +112,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentTimest
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowCurrentUser;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDataNodes;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowFunctions;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowPipePlugins;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowPipes;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowRegions;
@@ -658,4 +665,22 @@ public class TableConfigTaskVisitor extends AstVisitor<IConfigTask, MPPQueryCont
context.setQueryType(QueryType.READ);
return new ShowCurrentTimestampTask();
}
+
+ @Override
+ protected IConfigTask visitCreateFunction(CreateFunction node, MPPQueryContext context) {
+ context.setQueryType(QueryType.WRITE);
+ return new CreateFunctionTask(node);
+ }
+
+ @Override
+ protected IConfigTask visitShowFunctions(ShowFunctions node, MPPQueryContext context) {
+ context.setQueryType(QueryType.WRITE);
+ return new ShowFunctionsTask(Model.TABLE);
+ }
+
+ @Override
+ protected IConfigTask visitDropFunction(DropFunction node, MPPQueryContext context) {
+ context.setQueryType(QueryType.WRITE);
+ return new DropFunctionTask(Model.TABLE, node.getUdfName());
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
index ea96580b5b6..89b053ac475 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
import org.apache.iotdb.commons.partition.SchemaPartition;
import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.udf.service.UDFManagementService;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
@@ -46,6 +47,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.type.TypeSignature;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.TableBuiltinScalarFunction;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.utils.constant.SqlConstant;
+import org.apache.iotdb.udf.api.relational.ScalarFunction;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.read.common.type.BlobType;
@@ -618,6 +620,10 @@ public class TableMetadataImpl implements Metadata {
}
// TODO scalar UDF function
+ // 根据 argumentTypes 获取返回类型，这边暂时先 mock 一个 INT32
+ if (UDFManagementService.getInstance().isAssignableFrom(functionName, ScalarFunction.class)) {
+ return INT32;
+ }
// TODO UDAF
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java
index eb5878433a2..f2af9d7ea9c 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java
@@ -46,6 +46,37 @@ public class UDFDataTypeTransformer {
.collect(Collectors.toList());
}
+ public static TSDataType transformReadTypeToTSDataType(
+ org.apache.tsfile.read.common.type.Type type) {
+ if (type == null) {
+ return null;
+ }
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ return TSDataType.BOOLEAN;
+ case INT32:
+ return TSDataType.INT32;
+ case INT64:
+ return TSDataType.INT64;
+ case FLOAT:
+ return TSDataType.FLOAT;
+ case DOUBLE:
+ return TSDataType.DOUBLE;
+ case TEXT:
+ return TSDataType.TEXT;
+ case TIMESTAMP:
+ return TSDataType.TIMESTAMP;
+ case DATE:
+ return TSDataType.DATE;
+ case BLOB:
+ return TSDataType.BLOB;
+ case STRING:
+ return TSDataType.STRING;
+ default:
+ throw new IllegalArgumentException("Invalid input: " + type);
+ }
+ }
+
private static Type getUDFDataType(byte type) {
switch (type) {
case 0:
(iotdb) 03/06: save
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a commit to branch udsf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 499ae65fc08a64d296b00c7b9ef4f0182184f132
Author: Chen YZ <ch...@gmail.com>
AuthorDate: Tue Nov 19 16:06:07 2024 +0800
save
---
iotdb-api/udf-api/pom.xml | 6 +
.../config/ScalarFunctionConfig.java} | 12 +-
.../customizer/parameter/FunctionParameters.java | 56 +++++++++
...egationFunction.java => AggregateFunction.java} | 2 +-
.../iotdb/udf/api/relational/ScalarFunction.java | 30 ++++-
.../execution/aggregation/UDAFAccumulator.java | 3 +-
.../relational/metadata/TableMetadataImpl.java | 21 +++-
.../udf/UserDefineScalarFunctionTransformer.java | 125 +++++++++++++++++++++
.../transformation/dag/udf/UDTFExecutor.java | 3 +-
.../udf/builtin/UDTFEqualSizeBucketSample.java | 3 +-
.../apache/iotdb/commons/udf/builtin/UDTFJexl.java | 3 +-
.../apache/iotdb/commons/udf/builtin/UDTFM4.java | 6 +-
.../iotdb/commons/udf/utils/TableUDFUtils.java | 54 +++++++++
.../commons/udf/utils/UDFDataTypeTransformer.java | 62 ++++++++++
14 files changed, 373 insertions(+), 13 deletions(-)
diff --git a/iotdb-api/udf-api/pom.xml b/iotdb-api/udf-api/pom.xml
index 8a66274237f..734b7a5fb0c 100644
--- a/iotdb-api/udf-api/pom.xml
+++ b/iotdb-api/udf-api/pom.xml
@@ -34,6 +34,12 @@
<artifactId>common</artifactId>
<version>${tsfile.version}</version>
</dependency>
+ <dependency>
+ <groupId>net.bytebuddy</groupId>
+ <artifactId>byte-buddy</artifactId>
+ <version>1.15.10</version>
+ <scope>compile</scope>
+ </dependency>
</dependencies>
<profiles>
<profile>
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/AggregationFunction.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/config/ScalarFunctionConfig.java
similarity index 73%
copy from iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/AggregationFunction.java
copy to iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/config/ScalarFunctionConfig.java
index e831d5b5292..adacabed085 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/AggregationFunction.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/config/ScalarFunctionConfig.java
@@ -17,6 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.udf.api.relational;
+package org.apache.iotdb.udf.api.customizer.config;
-public interface AggregationFunction extends SQLFunction {}
+import org.apache.iotdb.udf.api.type.Type;
+
+public class ScalarFunctionConfig extends UDFConfigurations {
+
+ public ScalarFunctionConfig setOutputDataType(Type outputDataType) {
+ this.outputDataType = outputDataType;
+ return this;
+ }
+}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/parameter/FunctionParameters.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/parameter/FunctionParameters.java
new file mode 100644
index 00000000000..5af832cbaa7
--- /dev/null
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/customizer/parameter/FunctionParameters.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.iotdb.udf.api.customizer.parameter;
+
+import org.apache.iotdb.udf.api.type.Type;
+
+import java.util.List;
+import java.util.Map;
+
+public class FunctionParameters {
+ private final List<Type> childExpressionDataTypes;
+ private final Map<String, String> systemAttributes;
+
+ public FunctionParameters(
+ List<Type> childExpressionDataTypes, Map<String, String> systemAttributes) {
+ this.childExpressionDataTypes = childExpressionDataTypes;
+ this.systemAttributes = systemAttributes;
+ }
+
+ public List<Type> getChildExpressionDataTypes() {
+ return childExpressionDataTypes;
+ }
+
+ public int getChildExpressionsSize() {
+ return childExpressionDataTypes.size();
+ }
+
+ public Type getDataType(int index) {
+ return childExpressionDataTypes.get(index);
+ }
+
+ public boolean hasSystemAttribute(String attributeKey) {
+ return systemAttributes.containsKey(attributeKey);
+ }
+
+ public Map<String, String> getSystemAttributes() {
+ return systemAttributes;
+ }
+}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/AggregationFunction.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/AggregateFunction.java
similarity index 93%
rename from iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/AggregationFunction.java
rename to iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/AggregateFunction.java
index e831d5b5292..24942afd010 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/AggregationFunction.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/AggregateFunction.java
@@ -19,4 +19,4 @@
package org.apache.iotdb.udf.api.relational;
-public interface AggregationFunction extends SQLFunction {}
+public interface AggregateFunction extends SQLFunction {}
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/ScalarFunction.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/ScalarFunction.java
index 996d6994138..bd0586d051c 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/ScalarFunction.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/ScalarFunction.java
@@ -19,4 +19,32 @@
package org.apache.iotdb.udf.api.relational;
-public interface ScalarFunction extends SQLFunction {}
+import org.apache.iotdb.udf.api.customizer.parameter.FunctionParameters;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.type.Type;
+
+public interface ScalarFunction extends SQLFunction {
+
+ /**
+ * This method is mainly used to validate {@link FunctionParameters} and infer output data type.
+ *
+ * @param parameters parameters used to validate
+ * @throws Exception if any parameter is not valid
+ */
+ Type validateAndInferOutputType(FunctionParameters parameters) throws Exception;
+
+ /**
+ * This method will be called to process the transformation. In a single UDF query, this method
+ * may be called multiple times.
+ *
+ * @param input original input data row
+ * @throws Exception the user can throw errors if necessary
+ * @throws UnsupportedOperationException if the user does not override this method
+ */
+ Object evaluate(Record input) throws Exception;
+
+ /** This method is mainly used to release the resources used in the SQLFunction. */
+ default void beforeDestroy() {
+ // do nothing
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/UDAFAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/UDAFAccumulator.java
index a0b78cc8cd6..a56d6e9bd03 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/UDAFAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/UDAFAccumulator.java
@@ -199,7 +199,8 @@ public class UDAFAccumulator implements Accumulator {
@Override
public TSDataType getFinalType() {
- return UDFDataTypeTransformer.transformToTsDataType(configurations.getOutputDataType());
+ return UDFDataTypeTransformer.transformToTsDataType(
+ configurations.getOutputDataType());
}
private void onError(String methodName, Exception e) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
index 89b053ac475..e127de7186d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
import org.apache.iotdb.commons.partition.SchemaPartition;
import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.commons.udf.service.UDFManagementService;
+import org.apache.iotdb.commons.udf.utils.TableUDFUtils;
+import org.apache.iotdb.commons.udf.utils.UDFDataTypeTransformer;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
@@ -47,6 +49,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.type.TypeSignature;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.TableBuiltinScalarFunction;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.utils.constant.SqlConstant;
+import org.apache.iotdb.udf.api.customizer.parameter.FunctionParameters;
import org.apache.iotdb.udf.api.relational.ScalarFunction;
import org.apache.tsfile.file.metadata.IDeviceID;
@@ -619,10 +622,22 @@ public class TableMetadataImpl implements Metadata {
// ignore
}
- // TODO scalar UDF function
// 根据 argumentTypes 获取返回类型，这边暂时先 mock 一个 INT32
- if (UDFManagementService.getInstance().isAssignableFrom(functionName, ScalarFunction.class)) {
- return INT32;
+ if (TableUDFUtils.isScalarFunction(functionName)) {
+ ScalarFunction scalarFunction =
+ UDFManagementService.getInstance().reflect(functionName, ScalarFunction.class);
+ FunctionParameters functionParameters =
+ new FunctionParameters(
+ argumentTypes.stream()
+ .map(UDFDataTypeTransformer::transformReadTypeToUDFDataType)
+ .collect(Collectors.toList()),
+ Collections.emptyMap());
+ try {
+ return UDFDataTypeTransformer.transformUDFDataTypeToReadType(
+ scalarFunction.validateAndInferOutputType(functionParameters));
+ } catch (Exception e) {
+ throw new SemanticException("Invalid function parameters: " + e.getMessage());
+ }
}
// TODO UDAF
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java
new file mode 100644
index 00000000000..7689bbcca9d
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.iotdb.db.queryengine.transformation.dag.column.udf;
+
+import org.apache.iotdb.commons.udf.service.UDFManagementService;
+import org.apache.iotdb.commons.udf.utils.UDFDataTypeTransformer;
+import org.apache.iotdb.db.exception.sql.SemanticException;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.ColumnTransformer;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.multi.MultiColumnTransformer;
+import org.apache.iotdb.db.queryengine.transformation.dag.udf.UDFParametersFactory;
+import org.apache.iotdb.udf.api.access.ColumnToRowIterator;
+import org.apache.iotdb.udf.api.customizer.parameter.UDFParameters;
+import org.apache.iotdb.udf.api.relational.ScalarFunction;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.type.Type;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+// TODO(UDSF): encapsulate refect and validate logic
+public class UserDefineScalarFunctionTransformer extends MultiColumnTransformer {
+
+ private final ScalarFunction scalarFunction;
+ private final List<TSDataType> childrenTypes;
+
+ public UserDefineScalarFunctionTransformer(
+ Type returnType,
+ String functionName,
+ List<Expression> children,
+ List<ColumnTransformer> childrenTransformers) {
+ super(returnType, childrenTransformers);
+ ScalarFunction scalarFunction =
+ UDFManagementService.getInstance().reflect(functionName, ScalarFunction.class);
+ this.childrenTypes =
+ childrenTransformers.stream()
+ .map(ColumnTransformer::getType)
+ .map(UDFDataTypeTransformer::transformReadTypeToTSDataType)
+ .collect(Collectors.toList());
+ // TODO: 1、Table UDF 里不应该再用 String Expression 了
+ // TODO：2、想办法弄到 attributes
+ UDFParameters udfParameters =
+ UDFParametersFactory.buildUdfParameters(
+ children.stream().map(Expression::toString).collect(Collectors.toList()),
+ childrenTypes,
+ Collections.emptyMap());
+ try {
+ // scalarFunction.validate(new UDFParameterValidator(udfParameters));
+ // scalarFunction.beforeStart(udfParameters, new ScalarFunctionConfig());
+ } catch (Exception e) {
+ throw new SemanticException(e.getMessage());
+ }
+
+ this.scalarFunction = scalarFunction;
+ }
+
+ @Override
+ protected void doTransform(
+ List<Column> childrenColumns, ColumnBuilder builder, int positionCount) {
+ ColumnToRowIterator iterator =
+ new ColumnToRowIterator(childrenTypes, childrenColumns, positionCount);
+ // while (iterator.hasNextRow()) {
+ // try {
+ // Row row = iterator.next();
+ // Object result = scalarFunction.evaluate(row);
+ // if (result == null) {
+ // builder.appendNull();
+ // } else {
+ // builder.writeObject(result);
+ // }
+ // } catch (Exception e) {
+ // throw new RuntimeException(
+ // "Error occurs when evaluating UDF " + scalarFunction.getClass().getName(), e);
+ // }
+ // }
+ }
+
+ @Override
+ protected void doTransform(
+ List<Column> childrenColumns, ColumnBuilder builder, int positionCount, boolean[] selection) {
+ ColumnToRowIterator iterator =
+ new ColumnToRowIterator(childrenTypes, childrenColumns, positionCount);
+ int i = 0;
+ // while (iterator.hasNextRow()) {
+ // try {
+ // Row row = iterator.next();
+ // Object result = scalarFunction.evaluate(row);
+ // if (selection[i++] || result == null) {
+ // builder.appendNull();
+ // } else {
+ // builder.writeObject(result);
+ // }
+ // } catch (Exception e) {
+ // throw new RuntimeException(
+ // "Error occurs when evaluating UDF " + scalarFunction.getClass().getName(), e);
+ // }
+ // }
+ }
+
+ @Override
+ protected void checkType() {
+ // TODO: implement this method
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/udf/UDTFExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/udf/UDTFExecutor.java
index f93de5c1dc4..a9ab775360b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/udf/UDTFExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/udf/UDTFExecutor.java
@@ -80,7 +80,8 @@ public class UDTFExecutor {
configurations.getAccessStrategy().getAccessStrategyType())) {
outputStorage =
ElasticSerializableTVList.construct(
- UDFDataTypeTransformer.transformToTsDataType(configurations.getOutputDataType()),
+ UDFDataTypeTransformer.transformToTsDataType(
+ configurations.getOutputDataType()),
queryId,
collectorMemoryBudgetInMB,
1);
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFEqualSizeBucketSample.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFEqualSizeBucketSample.java
index 306d35e110f..d3492ecbe2e 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFEqualSizeBucketSample.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFEqualSizeBucketSample.java
@@ -45,7 +45,8 @@ public abstract class UDTFEqualSizeBucketSample implements UDTF {
"Illegal sample proportion. proportion > 0 and proportion <= 1",
proportion);
dataType =
- UDFDataTypeTransformer.transformToTsDataType(validator.getParameters().getDataType(0));
+ UDFDataTypeTransformer.transformToTsDataType(
+ validator.getParameters().getDataType(0));
bucketSize = (int) (1 / proportion);
}
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFJexl.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFJexl.java
index d554cfe2572..404ab2f26e7 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFJexl.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFJexl.java
@@ -70,7 +70,8 @@ public class UDTFJexl implements UDTF {
inputDataType = new TSDataType[inputSeriesNumber];
for (int i = 0; i < inputSeriesNumber; i++) {
- inputDataType[i] = UDFDataTypeTransformer.transformToTsDataType(parameters.getDataType(i));
+ inputDataType[i] =
+ UDFDataTypeTransformer.transformToTsDataType(parameters.getDataType(i));
}
outputDataType = probeOutputDataType();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFM4.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFM4.java
index 824740b8c9b..ccbb2848a01 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFM4.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFM4.java
@@ -94,14 +94,16 @@ public class UDTFM4 implements UDTF {
}
dataType =
- UDFDataTypeTransformer.transformToTsDataType(validator.getParameters().getDataType(0));
+ UDFDataTypeTransformer.transformToTsDataType(
+ validator.getParameters().getDataType(0));
}
@Override
public void beforeStart(UDFParameters parameters, UDTFConfigurations configurations)
throws MetadataException {
// set data type
- configurations.setOutputDataType(UDFDataTypeTransformer.transformToUDFDataType(dataType));
+ configurations.setOutputDataType(
+ UDFDataTypeTransformer.transformToUDFDataType(dataType));
// set access strategy
if (accessStrategy == AccessStrategy.SIZE_WINDOW) {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/TableUDFUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/TableUDFUtils.java
new file mode 100644
index 00000000000..bc2d42507a9
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/TableUDFUtils.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.iotdb.commons.udf.utils;
+
+import org.apache.iotdb.commons.udf.service.UDFManagementService;
+import org.apache.iotdb.udf.api.relational.AggregateFunction;
+import org.apache.iotdb.udf.api.relational.ScalarFunction;
+import org.apache.iotdb.udf.api.relational.TableFunction;
+
+public class TableUDFUtils {
+ public static boolean isScalarFunction(String functionName) {
+ try {
+ UDFManagementService.getInstance().reflect(functionName, ScalarFunction.class);
+ return true;
+ } catch (Throwable e) {
+ return false;
+ }
+ }
+
+ public static boolean isTableFunction(String functionName) {
+ try {
+ UDFManagementService.getInstance().reflect(functionName, TableFunction.class);
+ return true;
+ } catch (Throwable e) {
+ return false;
+ }
+ }
+
+ public static boolean isAggregateFunction(String functionName) {
+ try {
+ UDFManagementService.getInstance().reflect(functionName, AggregateFunction.class);
+ return true;
+ } catch (Throwable e) {
+ return false;
+ }
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java
index f2af9d7ea9c..1cae52ea5b9 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/UDFDataTypeTransformer.java
@@ -21,6 +21,12 @@ package org.apache.iotdb.commons.udf.utils;
import org.apache.iotdb.udf.api.type.Type;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.type.BinaryType;
+import org.apache.tsfile.read.common.type.BooleanType;
+import org.apache.tsfile.read.common.type.DoubleType;
+import org.apache.tsfile.read.common.type.FloatType;
+import org.apache.tsfile.read.common.type.IntType;
+import org.apache.tsfile.read.common.type.LongType;
import java.util.List;
import java.util.stream.Collectors;
@@ -77,6 +83,62 @@ public class UDFDataTypeTransformer {
}
}
+ public static Type transformReadTypeToUDFDataType(org.apache.tsfile.read.common.type.Type type) {
+ if (type == null) {
+ return null;
+ }
+ switch (type.getTypeEnum()) {
+ case BOOLEAN:
+ return Type.BOOLEAN;
+ case INT32:
+ return Type.INT32;
+ case INT64:
+ return Type.INT64;
+ case FLOAT:
+ return Type.FLOAT;
+ case DOUBLE:
+ return Type.DOUBLE;
+ case TEXT:
+ return Type.TEXT;
+ case TIMESTAMP:
+ return Type.TIMESTAMP;
+ case DATE:
+ return Type.DATE;
+ case BLOB:
+ return Type.BLOB;
+ case STRING:
+ return Type.STRING;
+ default:
+ throw new IllegalArgumentException("Invalid input: " + type);
+ }
+ }
+
+ public static org.apache.tsfile.read.common.type.Type transformUDFDataTypeToReadType(Type type) {
+ if (type == null) {
+ return null;
+ }
+ switch (type) {
+ case BOOLEAN:
+ return BooleanType.BOOLEAN;
+ case INT32:
+ case DATE:
+ return IntType.INT32;
+ case INT64:
+ case TIMESTAMP:
+ return LongType.INT64;
+ case FLOAT:
+ return FloatType.FLOAT;
+ case DOUBLE:
+ return DoubleType.DOUBLE;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ return BinaryType.TEXT;
+ default:
+ throw new IllegalArgumentException("Invalid input: " + type);
+ }
+ }
+
private static Type getUDFDataType(byte type) {
switch (type) {
case 0:
(iotdb) 06/06: fix it
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a commit to branch udsf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c215e63e0bedd5d80318d8e05814b40f0fcb5efc
Author: Chen YZ <ch...@gmail.com>
AuthorDate: Wed Nov 20 11:10:44 2024 +0800
fix it
---
.../apache/iotdb/udf/ScalarFunctionExample.java | 44 ++++++++++
.../confignode/it/IoTDBConfigNodeSnapshotIT.java | 2 +
.../iotdb/udf/api/relational/access/Record.java | 9 ++
.../execution/aggregation/UDAFAccumulator.java | 3 +-
.../execution/config/TableConfigTaskVisitor.java | 2 +-
.../config/executor/ClusterConfigTaskExecutor.java | 12 +--
.../config/metadata/ShowFunctionsTask.java | 96 +++++++++++++++++++++-
.../plan/relational/sql/parser/AstBuilder.java | 5 +-
.../unary/scalar/TableBuiltinScalarFunction.java | 15 ++++
.../transformation/dag/udf/UDTFExecutor.java | 3 +-
.../apache/iotdb/commons/conf/IoTDBConstant.java | 11 +++
.../iotdb/commons/udf/access/RecordIterator.java | 5 ++
.../udf/builtin/UDTFEqualSizeBucketSample.java | 3 +-
.../apache/iotdb/commons/udf/builtin/UDTFJexl.java | 3 +-
.../apache/iotdb/commons/udf/builtin/UDTFM4.java | 6 +-
.../commons/udf/service/UDFManagementService.java | 5 +-
16 files changed, 197 insertions(+), 27 deletions(-)
diff --git a/example/udf/src/main/java/org/apache/iotdb/udf/ScalarFunctionExample.java b/example/udf/src/main/java/org/apache/iotdb/udf/ScalarFunctionExample.java
new file mode 100644
index 00000000000..b0718a5fc20
--- /dev/null
+++ b/example/udf/src/main/java/org/apache/iotdb/udf/ScalarFunctionExample.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.iotdb.udf;
+
+import org.apache.iotdb.udf.api.customizer.parameter.FunctionParameters;
+import org.apache.iotdb.udf.api.relational.ScalarFunction;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.type.Type;
+
+public class ScalarFunctionExample implements ScalarFunction {
+ @Override
+ public void validate(FunctionParameters parameters) throws Exception {
+ if (parameters.getChildExpressionsSize() > 1) {
+ throw new Exception("Only one child expression is allowed");
+ }
+ }
+
+ @Override
+ public Type inferOutputType(FunctionParameters parameters) {
+ return parameters.getChildExpressionDataTypes().get(0);
+ }
+
+ @Override
+ public Object evaluate(Record input) throws Exception {
+ return input.getObject(0);
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java
index 7a9d99006a8..046f3418c0f 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java
@@ -285,12 +285,14 @@ public class IoTDBConfigNodeSnapshotIT {
TCreateFunctionReq createFunctionReq1 =
new TCreateFunctionReq("test1", "org.apache.iotdb.udf.UDTFExample", true)
+ .setModel(Model.TREE)
.setJarName(jarName)
.setJarFile(jarFile)
.setJarMD5(jarMD5);
TCreateFunctionReq createFunctionReq2 =
new TCreateFunctionReq("test2", "org.apache.iotdb.udf.UDTFExample", true)
+ .setModel(Model.TREE)
.setJarName(jarName)
.setJarFile(jarFile)
.setJarMD5(jarMD5);
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/access/Record.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/access/Record.java
index 98d3b6b80c5..e230b8bed37 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/access/Record.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/access/Record.java
@@ -104,6 +104,15 @@ public interface Record {
*/
String getString(int columnIndex) throws IOException;
+ /**
+ * Returns the object value at the specified column in this row.
+ *
+ * @param columnIndex index of the specified column
+ * @return the object value at the specified column in this row
+ * @throws IOException if an I/O error occurs
+ */
+ Object getObject(int columnIndex) throws IOException;
+
/**
* Returns the actual data type of the value at the specified column in this row.
*
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/UDAFAccumulator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/UDAFAccumulator.java
index a56d6e9bd03..a0b78cc8cd6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/UDAFAccumulator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/UDAFAccumulator.java
@@ -199,8 +199,7 @@ public class UDAFAccumulator implements Accumulator {
@Override
public TSDataType getFinalType() {
- return UDFDataTypeTransformer.transformToTsDataType(
- configurations.getOutputDataType());
+ return UDFDataTypeTransformer.transformToTsDataType(configurations.getOutputDataType());
}
private void onError(String methodName, Exception e) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
index cc514c7c04f..3f74ca1b988 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
@@ -674,7 +674,7 @@ public class TableConfigTaskVisitor extends AstVisitor<IConfigTask, MPPQueryCont
@Override
protected IConfigTask visitShowFunctions(ShowFunctions node, MPPQueryContext context) {
- context.setQueryType(QueryType.WRITE);
+ context.setQueryType(QueryType.READ);
return new ShowFunctionsTask(Model.TABLE);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 82476db718a..68421b93094 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -628,8 +628,8 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
getUDFTableResp.getStatus().message, getUDFTableResp.getStatus().code));
return future;
}
- // convert UDFTable and buildTsBlock
- ShowFunctionsTask.buildTsBlock(getUDFTableResp.getAllUDFInformation(), future);
+ // convert UDFTable and buildTreeModelTsBlock
+ ShowFunctionsTask.buildTsBlock(model, getUDFTableResp.getAllUDFInformation(), future);
} catch (ClientManagerException | TException e) {
future.setException(e);
}
@@ -803,7 +803,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
getTriggerTableResp.getStatus().message, getTriggerTableResp.getStatus().code));
return future;
}
- // convert triggerTable and buildTsBlock
+ // convert triggerTable and buildTreeModelTsBlock
ShowTriggersTask.buildTsBlock(getTriggerTableResp.getAllTriggerInformation(), future);
} catch (ClientManagerException | TException e) {
future.setException(e);
@@ -978,7 +978,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
getPipePluginTableResp.getStatus().code));
return future;
}
- // convert PipePluginTable and buildTsBlock
+ // convert PipePluginTable and buildTreeModelTsBlock
ShowPipePluginsTask.buildTsBlock(getPipePluginTableResp.getAllPipePluginMeta(), future);
} catch (ClientManagerException | TException e) {
future.setException(e);
@@ -2826,7 +2826,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
new IoTDBException(showCQResp.getStatus().message, showCQResp.getStatus().code));
return future;
}
- // convert cqList and buildTsBlock
+ // convert cqList and buildTreeModelTsBlock
ShowContinuousQueriesTask.buildTsBlock(showCQResp.getCqList(), future);
} catch (ClientManagerException | TException e) {
future.setException(e);
@@ -2893,7 +2893,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
new IoTDBException(showModelResp.getStatus().message, showModelResp.getStatus().code));
return future;
}
- // convert model info list and buildTsBlock
+ // convert model info list and buildTreeModelTsBlock
ShowModelsTask.buildTsBlock(showModelResp.getModelInfoList(), future);
} catch (ClientManagerException | TException e) {
future.setException(e);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowFunctionsTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowFunctionsTask.java
index a0a8506d0b6..3295b20350f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowFunctionsTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/ShowFunctionsTask.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.udf.UDFType;
import org.apache.iotdb.commons.udf.builtin.BuiltinAggregationFunction;
import org.apache.iotdb.commons.udf.builtin.BuiltinScalarFunction;
import org.apache.iotdb.commons.udf.builtin.BuiltinTimeSeriesGeneratingFunction;
+import org.apache.iotdb.commons.udf.utils.TableUDFUtils;
import org.apache.iotdb.commons.udf.utils.TreeUDFUtils;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
@@ -33,6 +34,8 @@ import org.apache.iotdb.db.queryengine.common.header.DatasetHeaderFactory;
import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.TableBuiltinAggregationFunction;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.TableBuiltinScalarFunction;
import org.apache.iotdb.rpc.TSStatusCode;
import com.google.common.util.concurrent.ListenableFuture;
@@ -44,6 +47,7 @@ import org.apache.tsfile.utils.BytesUtils;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
@@ -52,12 +56,18 @@ import java.util.stream.Collectors;
import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_STATE_AVAILABLE;
import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_STATE_UNAVAILABLE;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_BUILTIN_AGG_FUNC;
import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_BUILTIN_SCALAR;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_BUILTIN_SCALAR_FUNC;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_BUILTIN_TABLE_FUNC;
import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_BUILTIN_UDTF;
import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_EXTERNAL_UDAF;
import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_EXTERNAL_UDTF;
import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_NATIVE;
import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_UNKNOWN;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_USER_DEFINED_AGG_FUNC;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_USER_DEFINED_SCALAR_FUNC;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.FUNCTION_TYPE_USER_DEFINED_TABLE_FUNC;
public class ShowFunctionsTask implements IConfigTask {
@@ -69,9 +79,27 @@ public class ShowFunctionsTask implements IConfigTask {
BINARY_MAP.put(FUNCTION_TYPE_EXTERNAL_UDTF, BytesUtils.valueOf(FUNCTION_TYPE_EXTERNAL_UDTF));
BINARY_MAP.put(FUNCTION_TYPE_EXTERNAL_UDAF, BytesUtils.valueOf(FUNCTION_TYPE_EXTERNAL_UDAF));
BINARY_MAP.put(FUNCTION_TYPE_BUILTIN_SCALAR, BytesUtils.valueOf(FUNCTION_TYPE_BUILTIN_SCALAR));
+
+ BINARY_MAP.put(
+ FUNCTION_TYPE_BUILTIN_SCALAR_FUNC, BytesUtils.valueOf(FUNCTION_TYPE_BUILTIN_SCALAR_FUNC));
+ BINARY_MAP.put(
+ FUNCTION_TYPE_BUILTIN_AGG_FUNC, BytesUtils.valueOf(FUNCTION_TYPE_BUILTIN_AGG_FUNC));
+ BINARY_MAP.put(
+ FUNCTION_TYPE_BUILTIN_TABLE_FUNC, BytesUtils.valueOf(FUNCTION_TYPE_BUILTIN_TABLE_FUNC));
+ BINARY_MAP.put(
+ FUNCTION_TYPE_USER_DEFINED_SCALAR_FUNC,
+ BytesUtils.valueOf(FUNCTION_TYPE_USER_DEFINED_SCALAR_FUNC));
+ BINARY_MAP.put(
+ FUNCTION_TYPE_USER_DEFINED_AGG_FUNC,
+ BytesUtils.valueOf(FUNCTION_TYPE_USER_DEFINED_AGG_FUNC));
+ BINARY_MAP.put(
+ FUNCTION_TYPE_USER_DEFINED_TABLE_FUNC,
+ BytesUtils.valueOf(FUNCTION_TYPE_USER_DEFINED_TABLE_FUNC));
+
BINARY_MAP.put(FUNCTION_TYPE_UNKNOWN, BytesUtils.valueOf(FUNCTION_TYPE_UNKNOWN));
BINARY_MAP.put(FUNCTION_STATE_AVAILABLE, BytesUtils.valueOf(FUNCTION_STATE_AVAILABLE));
BINARY_MAP.put(FUNCTION_STATE_UNAVAILABLE, BytesUtils.valueOf(FUNCTION_STATE_UNAVAILABLE));
+
BINARY_MAP.put("", BytesUtils.valueOf(""));
}
@@ -87,6 +115,48 @@ public class ShowFunctionsTask implements IConfigTask {
}
public static void buildTsBlock(
+ Model model, List<ByteBuffer> allUDFInformation, SettableFuture<ConfigTaskResult> future) {
+ if (Model.TREE.equals(model)) {
+ buildTreeModelTsBlock(allUDFInformation, future);
+ } else {
+ buildTableModelTsBlock(allUDFInformation, future);
+ }
+ }
+
+ private static void buildTableModelTsBlock(
+ List<ByteBuffer> allUDFInformation, SettableFuture<ConfigTaskResult> future) {
+ List<TSDataType> outputDataTypes =
+ ColumnHeaderConstant.showFunctionsColumnHeaders.stream()
+ .map(ColumnHeader::getColumnType)
+ .collect(Collectors.toList());
+ TsBlockBuilder builder = new TsBlockBuilder(outputDataTypes);
+ List<UDFInformation> udfInformations = new ArrayList<>();
+ if (allUDFInformation != null && !allUDFInformation.isEmpty()) {
+ for (ByteBuffer udfInformationByteBuffer : allUDFInformation) {
+ UDFInformation udfInformation = UDFInformation.deserialize(udfInformationByteBuffer);
+ udfInformations.add(udfInformation);
+ }
+ }
+
+ udfInformations.sort(Comparator.comparing(UDFInformation::getFunctionName));
+ for (UDFInformation udfInformation : udfInformations) {
+ appendUDFInformation(builder, udfInformation);
+ }
+ appendFunctions(
+ builder,
+ TableBuiltinScalarFunction.getNativeFunctionNames(),
+ BINARY_MAP.get(FUNCTION_TYPE_BUILTIN_SCALAR_FUNC),
+ BINARY_MAP.get(FUNCTION_STATE_AVAILABLE));
+ appendFunctions(
+ builder,
+ TableBuiltinAggregationFunction.getNativeFunctionNames(),
+ BINARY_MAP.get(FUNCTION_TYPE_BUILTIN_AGG_FUNC),
+ BINARY_MAP.get(FUNCTION_STATE_AVAILABLE));
+ DatasetHeader datasetHeader = DatasetHeaderFactory.getShowFunctionsHeader();
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS, builder.build(), datasetHeader));
+ }
+
+ private static void buildTreeModelTsBlock(
List<ByteBuffer> allUDFInformation, SettableFuture<ConfigTaskResult> future) {
List<TSDataType> outputDataTypes =
ColumnHeaderConstant.showFunctionsColumnHeaders.stream()
@@ -118,10 +188,25 @@ public class ShowFunctionsTask implements IConfigTask {
builder.getColumnBuilder(1).writeBinary(getFunctionType(udfInformation));
builder.getColumnBuilder(2).writeBinary(BytesUtils.valueOf(udfInformation.getClassName()));
builder.getColumnBuilder(3).writeBinary(getFunctionState(udfInformation));
-
builder.declarePosition();
}
+ private static void appendFunctions(
+ TsBlockBuilder builder,
+ Collection<String> functionNames,
+ Binary functionType,
+ Binary functionState) {
+ final Binary className = BINARY_MAP.get("");
+ for (String functionName : functionNames) {
+ builder.getTimeColumnBuilder().writeLong(0L);
+ builder.getColumnBuilder(0).writeBinary(BytesUtils.valueOf(functionName.toUpperCase()));
+ builder.getColumnBuilder(1).writeBinary(functionType);
+ builder.getColumnBuilder(2).writeBinary(className);
+ builder.getColumnBuilder(3).writeBinary(functionState);
+ builder.declarePosition();
+ }
+ }
+
private static void appendBuiltInTimeSeriesGeneratingFunctions(TsBlockBuilder builder) {
final Binary functionType = BINARY_MAP.get(FUNCTION_TYPE_BUILTIN_UDTF);
final Binary functionState = BINARY_MAP.get(FUNCTION_STATE_AVAILABLE);
@@ -175,6 +260,15 @@ public class ShowFunctionsTask implements IConfigTask {
} else if (TreeUDFUtils.isUDAF(udfInformation.getFunctionName())) {
return BINARY_MAP.get(FUNCTION_TYPE_EXTERNAL_UDAF);
}
+ } else {
+ if (TableUDFUtils.tryGetScalarFunction(udfInformation.getFunctionName()) != null) {
+ return BINARY_MAP.get(FUNCTION_TYPE_USER_DEFINED_SCALAR_FUNC);
+ } else if (TableUDFUtils.tryGetAggregateFunction(udfInformation.getFunctionName())
+ != null) {
+ return BINARY_MAP.get(FUNCTION_TYPE_USER_DEFINED_AGG_FUNC);
+ } else if (TableUDFUtils.tryGetTableFunction(udfInformation.getFunctionName()) != null) {
+ return BINARY_MAP.get(FUNCTION_TYPE_USER_DEFINED_TABLE_FUNC);
+ }
}
}
return BINARY_MAP.get(FUNCTION_TYPE_UNKNOWN);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
index c079a00439a..1a7237d4989 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
@@ -166,7 +166,6 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.util.AstUtil;
import org.apache.iotdb.db.queryengine.plan.statement.StatementType;
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement;
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement;
-import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateFunctionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.sys.FlushStatement;
import org.apache.iotdb.db.queryengine.plan.statement.sys.SetConfigurationStatement;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.TableBuiltinScalarFunction;
@@ -584,10 +583,10 @@ public class AstBuilder extends RelationalSqlBaseVisitor<Node> {
final String udfName = ((Identifier) visit(ctx.udfName)).getValue();
final String className = ((Identifier) visit(ctx.className)).getValue();
if (ctx.uriClause() == null) {
- return new CreateFunction(getLocation(ctx),udfName, className);
+ return new CreateFunction(getLocation(ctx), udfName, className);
} else {
String uriString = parseAndValidateURI(ctx.uriClause());
- return new CreateFunction(getLocation(ctx),udfName, className, uriString);
+ return new CreateFunction(getLocation(ctx), udfName, className, uriString);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/TableBuiltinScalarFunction.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/TableBuiltinScalarFunction.java
index c58182a3f27..237e4e4a793 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/TableBuiltinScalarFunction.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/unary/scalar/TableBuiltinScalarFunction.java
@@ -19,6 +19,11 @@
package org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+
public enum TableBuiltinScalarFunction {
DIFF("diff"),
CAST("cast"),
@@ -70,4 +75,14 @@ public enum TableBuiltinScalarFunction {
public String getFunctionName() {
return functionName;
}
+
+ private static final Set<String> NATIVE_FUNCTION_NAMES =
+ new HashSet<>(
+ Arrays.stream(TableBuiltinScalarFunction.values())
+ .map(TableBuiltinScalarFunction::getFunctionName)
+ .collect(Collectors.toList()));
+
+ public static Set<String> getNativeFunctionNames() {
+ return NATIVE_FUNCTION_NAMES;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/udf/UDTFExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/udf/UDTFExecutor.java
index a9ab775360b..f93de5c1dc4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/udf/UDTFExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/udf/UDTFExecutor.java
@@ -80,8 +80,7 @@ public class UDTFExecutor {
configurations.getAccessStrategy().getAccessStrategyType())) {
outputStorage =
ElasticSerializableTVList.construct(
- UDFDataTypeTransformer.transformToTsDataType(
- configurations.getOutputDataType()),
+ UDFDataTypeTransformer.transformToTsDataType(configurations.getOutputDataType()),
queryId,
collectorMemoryBudgetInMB,
1);
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
index b3de41b0664..cc6150f3400 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
@@ -178,12 +178,23 @@ public class IoTDBConstant {
public static final String COLUMN_SCHEMA_TEMPLATE = "template name";
+ // for tree model
public static final String FUNCTION_TYPE_NATIVE = "native";
public static final String FUNCTION_TYPE_BUILTIN_SCALAR = "built-in scalar";
public static final String FUNCTION_TYPE_BUILTIN_UDAF = "built-in UDAF";
public static final String FUNCTION_TYPE_BUILTIN_UDTF = "built-in UDTF";
public static final String FUNCTION_TYPE_EXTERNAL_UDAF = "external UDAF";
public static final String FUNCTION_TYPE_EXTERNAL_UDTF = "external UDTF";
+ // for table model
+ public static final String FUNCTION_TYPE_BUILTIN_SCALAR_FUNC = "built-in scalar function";
+ public static final String FUNCTION_TYPE_BUILTIN_AGG_FUNC = "built-in aggregate function";
+ public static final String FUNCTION_TYPE_BUILTIN_TABLE_FUNC = "built-in table function";
+ public static final String FUNCTION_TYPE_USER_DEFINED_SCALAR_FUNC =
+ "user-defined scalar function";
+ public static final String FUNCTION_TYPE_USER_DEFINED_AGG_FUNC =
+ "user-defined aggregate function";
+ public static final String FUNCTION_TYPE_USER_DEFINED_TABLE_FUNC = "user-defined table function";
+ // common
public static final String FUNCTION_TYPE_UNKNOWN = "UNKNOWN";
public static final String FUNCTION_STATE_AVAILABLE = "AVAILABLE";
public static final String FUNCTION_STATE_UNAVAILABLE = "UNAVAILABLE";
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/access/RecordIterator.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/access/RecordIterator.java
index 29f473b3c9d..49feb6d6b84 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/access/RecordIterator.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/access/RecordIterator.java
@@ -85,6 +85,11 @@ public class RecordIterator implements Iterator<Record> {
return childrenColumns.get(columnIndex).getBinary(index).toString();
}
+ @Override
+ public Object getObject(int columnIndex) throws IOException {
+ return childrenColumns.get(columnIndex).getObject(index);
+ }
+
@Override
public Type getDataType(int columnIndex) {
return UDFDataTypeTransformer.transformToUDFDataType(
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFEqualSizeBucketSample.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFEqualSizeBucketSample.java
index d3492ecbe2e..306d35e110f 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFEqualSizeBucketSample.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFEqualSizeBucketSample.java
@@ -45,8 +45,7 @@ public abstract class UDTFEqualSizeBucketSample implements UDTF {
"Illegal sample proportion. proportion > 0 and proportion <= 1",
proportion);
dataType =
- UDFDataTypeTransformer.transformToTsDataType(
- validator.getParameters().getDataType(0));
+ UDFDataTypeTransformer.transformToTsDataType(validator.getParameters().getDataType(0));
bucketSize = (int) (1 / proportion);
}
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFJexl.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFJexl.java
index 404ab2f26e7..d554cfe2572 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFJexl.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFJexl.java
@@ -70,8 +70,7 @@ public class UDTFJexl implements UDTF {
inputDataType = new TSDataType[inputSeriesNumber];
for (int i = 0; i < inputSeriesNumber; i++) {
- inputDataType[i] =
- UDFDataTypeTransformer.transformToTsDataType(parameters.getDataType(i));
+ inputDataType[i] = UDFDataTypeTransformer.transformToTsDataType(parameters.getDataType(i));
}
outputDataType = probeOutputDataType();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFM4.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFM4.java
index ccbb2848a01..824740b8c9b 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFM4.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/UDTFM4.java
@@ -94,16 +94,14 @@ public class UDTFM4 implements UDTF {
}
dataType =
- UDFDataTypeTransformer.transformToTsDataType(
- validator.getParameters().getDataType(0));
+ UDFDataTypeTransformer.transformToTsDataType(validator.getParameters().getDataType(0));
}
@Override
public void beforeStart(UDFParameters parameters, UDTFConfigurations configurations)
throws MetadataException {
// set data type
- configurations.setOutputDataType(
- UDFDataTypeTransformer.transformToUDFDataType(dataType));
+ configurations.setOutputDataType(UDFDataTypeTransformer.transformToUDFDataType(dataType));
// set access strategy
if (accessStrategy == AccessStrategy.SIZE_WINDOW) {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/service/UDFManagementService.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/service/UDFManagementService.java
index aa3e512294a..bbc7e2b436e 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/service/UDFManagementService.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/service/UDFManagementService.java
@@ -224,10 +224,7 @@ public class UDFManagementService {
try {
return clazz.cast(
- udfTable
- .getFunctionClass(Model.TREE, functionName)
- .getDeclaredConstructor()
- .newInstance());
+ udfTable.getFunctionClass(model, functionName).getDeclaredConstructor().newInstance());
} catch (InstantiationException
| InvocationTargetException
| NoSuchMethodException
(iotdb) 04/06: save
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chenyz pushed a commit to branch udsf
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b8a4fb6ab45adfbeb5f992b3ef4494795839c859
Author: Chen YZ <ch...@gmail.com>
AuthorDate: Tue Nov 19 16:43:56 2024 +0800
save
---
.../confignode/it/IoTDBConfigNodeSnapshotIT.java | 1 -
.../iotdb/udf/api/relational/ScalarFunction.java | 12 ++-
.../relational/ColumnTransformerBuilder.java | 32 ++++--
.../relational/metadata/TableMetadataImpl.java | 13 ++-
.../udf/UserDefineScalarFunctionTransformer.java | 109 ++++++++-------------
.../iotdb/commons/udf/access/RecordIterator.java | 105 ++++++++++++++++++++
.../iotdb/commons/udf/utils/TableUDFUtils.java | 21 ++--
7 files changed, 194 insertions(+), 99 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java
index 03faf93585c..7a9d99006a8 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java
@@ -31,7 +31,6 @@ import org.apache.iotdb.commons.path.PathDeserializeUtil;
import org.apache.iotdb.commons.trigger.TriggerInformation;
import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager;
import org.apache.iotdb.commons.udf.UDFInformation;
-import org.apache.iotdb.confignode.consensus.request.read.function.GetFunctionTablePlan;
import org.apache.iotdb.confignode.rpc.thrift.TCQEntry;
import org.apache.iotdb.confignode.rpc.thrift.TCreateCQReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
diff --git a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/ScalarFunction.java b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/ScalarFunction.java
index bd0586d051c..6f52103805e 100644
--- a/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/ScalarFunction.java
+++ b/iotdb-api/udf-api/src/main/java/org/apache/iotdb/udf/api/relational/ScalarFunction.java
@@ -26,12 +26,20 @@ import org.apache.iotdb.udf.api.type.Type;
public interface ScalarFunction extends SQLFunction {
/**
- * This method is mainly used to validate {@link FunctionParameters} and infer output data type.
+ * This method is used to validate {@link FunctionParameters}.
*
* @param parameters parameters used to validate
* @throws Exception if any parameter is not valid
*/
- Type validateAndInferOutputType(FunctionParameters parameters) throws Exception;
+ void validate(FunctionParameters parameters) throws Exception;
+
+ /**
+ * This method is used to infer the output data type of the transformation.
+ *
+ * @param parameters input parameters
+ * @return the output data type
+ */
+ Type inferOutputType(FunctionParameters parameters);
/**
* This method will be called to process the transformation. In a single UDF query, this method
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java
index 26b30bec4f8..429ffdeadac 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/relational/ColumnTransformerBuilder.java
@@ -19,7 +19,8 @@
package org.apache.iotdb.db.queryengine.execution.relational;
-import org.apache.iotdb.commons.udf.service.UDFManagementService;
+import org.apache.iotdb.commons.udf.utils.TableUDFUtils;
+import org.apache.iotdb.commons.udf.utils.UDFDataTypeTransformer;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
@@ -95,6 +96,7 @@ import org.apache.iotdb.db.queryengine.transformation.dag.column.multi.LogicalAn
import org.apache.iotdb.db.queryengine.transformation.dag.column.multi.LogicalOrMultiColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.ternary.BetweenColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.ternary.Like3ColumnTransformer;
+import org.apache.iotdb.db.queryengine.transformation.dag.column.udf.UserDefineScalarFunctionTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.IsNullColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.LikeColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.LogicNotColumnTransformer;
@@ -153,7 +155,7 @@ import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.Tr
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.TrimColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.TryCastFunctionColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.UpperColumnTransformer;
-import org.apache.iotdb.db.queryengine.transformation.dag.column.unary.scalar.UserDefineScalarFunctionTransformer;
+import org.apache.iotdb.udf.api.customizer.parameter.FunctionParameters;
import org.apache.iotdb.udf.api.relational.ScalarFunction;
import org.apache.tsfile.common.conf.TSFileConfig;
@@ -173,6 +175,7 @@ import org.apache.tsfile.utils.Binary;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -999,13 +1002,24 @@ public class ColumnTransformerBuilder
source,
((LongLiteral) children.get(3)).getParsedValue(),
context.sessionInfo.getZoneId());
- } else if (UDFManagementService.getInstance()
- .isAssignableFrom(functionName, ScalarFunction.class)) {
- List<ColumnTransformer> childrenColumnTransformer =
- children.stream().map(child -> process(child, context)).collect(Collectors.toList());
- // TODO(UDSF): check the return type of the function
- return new UserDefineScalarFunctionTransformer(
- INT32, functionName, children, childrenColumnTransformer);
+ } else {
+ // user defined function
+ ScalarFunction scalarFunction = TableUDFUtils.tryGetScalarFunction(functionName);
+ if (scalarFunction != null) {
+ List<ColumnTransformer> childrenColumnTransformer =
+ children.stream().map(child -> process(child, context)).collect(Collectors.toList());
+ FunctionParameters parameters =
+ new FunctionParameters(
+ childrenColumnTransformer.stream()
+ .map(i -> UDFDataTypeTransformer.transformReadTypeToUDFDataType(i.getType()))
+ .collect(Collectors.toList()),
+ Collections.emptyMap());
+ Type returnType =
+ UDFDataTypeTransformer.transformUDFDataTypeToReadType(
+ scalarFunction.inferOutputType(parameters));
+ return new UserDefineScalarFunctionTransformer(
+ returnType, scalarFunction, childrenColumnTransformer);
+ }
}
throw new IllegalArgumentException(String.format("Unknown function: %s", functionName));
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
index e127de7186d..d8c0c8c9b89 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
@@ -23,7 +23,6 @@ import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
import org.apache.iotdb.commons.partition.SchemaPartition;
import org.apache.iotdb.commons.schema.table.TsTable;
-import org.apache.iotdb.commons.udf.service.UDFManagementService;
import org.apache.iotdb.commons.udf.utils.TableUDFUtils;
import org.apache.iotdb.commons.udf.utils.UDFDataTypeTransformer;
import org.apache.iotdb.db.exception.sql.SemanticException;
@@ -622,10 +621,9 @@ public class TableMetadataImpl implements Metadata {
// ignore
}
- // 根据 argumentTypes 获取返回类型，这边暂时先 mock 一个 INT32
- if (TableUDFUtils.isScalarFunction(functionName)) {
- ScalarFunction scalarFunction =
- UDFManagementService.getInstance().reflect(functionName, ScalarFunction.class);
+ // User-defined scalar function
+ ScalarFunction scalarFunction = TableUDFUtils.tryGetScalarFunction(functionName);
+ if (scalarFunction != null) {
FunctionParameters functionParameters =
new FunctionParameters(
argumentTypes.stream()
@@ -633,11 +631,12 @@ public class TableMetadataImpl implements Metadata {
.collect(Collectors.toList()),
Collections.emptyMap());
try {
- return UDFDataTypeTransformer.transformUDFDataTypeToReadType(
- scalarFunction.validateAndInferOutputType(functionParameters));
+ scalarFunction.validate(functionParameters);
} catch (Exception e) {
throw new SemanticException("Invalid function parameters: " + e.getMessage());
}
+ return UDFDataTypeTransformer.transformUDFDataTypeToReadType(
+ scalarFunction.inferOutputType(functionParameters));
}
// TODO UDAF
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java
index 7689bbcca9d..c81c42135d5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/column/udf/UserDefineScalarFunctionTransformer.java
@@ -19,107 +19,80 @@
package org.apache.iotdb.db.queryengine.transformation.dag.column.udf;
-import org.apache.iotdb.commons.udf.service.UDFManagementService;
-import org.apache.iotdb.commons.udf.utils.UDFDataTypeTransformer;
-import org.apache.iotdb.db.exception.sql.SemanticException;
-import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.commons.udf.access.RecordIterator;
import org.apache.iotdb.db.queryengine.transformation.dag.column.ColumnTransformer;
import org.apache.iotdb.db.queryengine.transformation.dag.column.multi.MultiColumnTransformer;
-import org.apache.iotdb.db.queryengine.transformation.dag.udf.UDFParametersFactory;
-import org.apache.iotdb.udf.api.access.ColumnToRowIterator;
-import org.apache.iotdb.udf.api.customizer.parameter.UDFParameters;
import org.apache.iotdb.udf.api.relational.ScalarFunction;
+import org.apache.iotdb.udf.api.relational.access.Record;
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
-import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.type.Type;
-import java.util.Collections;
import java.util.List;
-import java.util.stream.Collectors;
-// TODO(UDSF): encapsulate refect and validate logic
public class UserDefineScalarFunctionTransformer extends MultiColumnTransformer {
private final ScalarFunction scalarFunction;
- private final List<TSDataType> childrenTypes;
public UserDefineScalarFunctionTransformer(
Type returnType,
- String functionName,
- List<Expression> children,
+ ScalarFunction scalarFunction,
List<ColumnTransformer> childrenTransformers) {
super(returnType, childrenTransformers);
- ScalarFunction scalarFunction =
- UDFManagementService.getInstance().reflect(functionName, ScalarFunction.class);
- this.childrenTypes =
- childrenTransformers.stream()
- .map(ColumnTransformer::getType)
- .map(UDFDataTypeTransformer::transformReadTypeToTSDataType)
- .collect(Collectors.toList());
- // TODO: 1、Table UDF 里不应该再用 String Expression 了
- // TODO：2、想办法弄到 attributes
- UDFParameters udfParameters =
- UDFParametersFactory.buildUdfParameters(
- children.stream().map(Expression::toString).collect(Collectors.toList()),
- childrenTypes,
- Collections.emptyMap());
- try {
- // scalarFunction.validate(new UDFParameterValidator(udfParameters));
- // scalarFunction.beforeStart(udfParameters, new ScalarFunctionConfig());
- } catch (Exception e) {
- throw new SemanticException(e.getMessage());
- }
-
this.scalarFunction = scalarFunction;
}
@Override
protected void doTransform(
List<Column> childrenColumns, ColumnBuilder builder, int positionCount) {
- ColumnToRowIterator iterator =
- new ColumnToRowIterator(childrenTypes, childrenColumns, positionCount);
- // while (iterator.hasNextRow()) {
- // try {
- // Row row = iterator.next();
- // Object result = scalarFunction.evaluate(row);
- // if (result == null) {
- // builder.appendNull();
- // } else {
- // builder.writeObject(result);
- // }
- // } catch (Exception e) {
- // throw new RuntimeException(
- // "Error occurs when evaluating UDF " + scalarFunction.getClass().getName(), e);
- // }
- // }
+ RecordIterator iterator = new RecordIterator(childrenColumns, positionCount);
+ while (iterator.hasNext()) {
+ try {
+ Object result = scalarFunction.evaluate(iterator.next());
+ if (result == null) {
+ builder.appendNull();
+ } else {
+ builder.writeObject(result);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Error occurs when evaluating user-defined scalar function "
+ + scalarFunction.getClass().getName(),
+ e);
+ }
+ }
}
@Override
protected void doTransform(
List<Column> childrenColumns, ColumnBuilder builder, int positionCount, boolean[] selection) {
- ColumnToRowIterator iterator =
- new ColumnToRowIterator(childrenTypes, childrenColumns, positionCount);
+ RecordIterator iterator = new RecordIterator(childrenColumns, positionCount);
int i = 0;
- // while (iterator.hasNextRow()) {
- // try {
- // Row row = iterator.next();
- // Object result = scalarFunction.evaluate(row);
- // if (selection[i++] || result == null) {
- // builder.appendNull();
- // } else {
- // builder.writeObject(result);
- // }
- // } catch (Exception e) {
- // throw new RuntimeException(
- // "Error occurs when evaluating UDF " + scalarFunction.getClass().getName(), e);
- // }
- // }
+ while (iterator.hasNext()) {
+ try {
+ Record input = iterator.next();
+ if (selection[i++]) {
+ builder.appendNull();
+ continue;
+ }
+ Object result = scalarFunction.evaluate(input);
+ if (result == null) {
+ builder.appendNull();
+ } else {
+ builder.writeObject(result);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Error occurs when evaluating user-defined scalar function "
+ + scalarFunction.getClass().getName(),
+ e);
+ }
+ }
}
@Override
protected void checkType() {
- // TODO: implement this method
+ // do nothing
}
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/access/RecordIterator.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/access/RecordIterator.java
new file mode 100644
index 00000000000..29f473b3c9d
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/access/RecordIterator.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.iotdb.commons.udf.access;
+
+import org.apache.iotdb.commons.udf.utils.UDFDataTypeTransformer;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.type.Binary;
+import org.apache.iotdb.udf.api.type.Type;
+
+import org.apache.tsfile.block.column.Column;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+
+public class RecordIterator implements Iterator<Record> {
+
+ private final List<Column> childrenColumns;
+ private final int positionCount;
+ private int currentIndex;
+
+ public RecordIterator(List<Column> childrenColumns, int positionCount) {
+ this.childrenColumns = childrenColumns;
+ this.positionCount = positionCount;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return currentIndex < positionCount;
+ }
+
+ @Override
+ public Record next() {
+ final int index = currentIndex++;
+ return new Record() {
+ @Override
+ public int getInt(int columnIndex) throws IOException {
+ return childrenColumns.get(columnIndex).getInt(index);
+ }
+
+ @Override
+ public long getLong(int columnIndex) throws IOException {
+ return childrenColumns.get(columnIndex).getLong(index);
+ }
+
+ @Override
+ public float getFloat(int columnIndex) throws IOException {
+ return childrenColumns.get(columnIndex).getFloat(index);
+ }
+
+ @Override
+ public double getDouble(int columnIndex) throws IOException {
+ return childrenColumns.get(columnIndex).getDouble(index);
+ }
+
+ @Override
+ public boolean getBoolean(int columnIndex) throws IOException {
+ return childrenColumns.get(columnIndex).getBoolean(index);
+ }
+
+ @Override
+ public Binary getBinary(int columnIndex) throws IOException {
+ return new Binary(childrenColumns.get(columnIndex).getBinary(index).getValues());
+ }
+
+ @Override
+ public String getString(int columnIndex) throws IOException {
+ return childrenColumns.get(columnIndex).getBinary(index).toString();
+ }
+
+ @Override
+ public Type getDataType(int columnIndex) {
+ return UDFDataTypeTransformer.transformToUDFDataType(
+ childrenColumns.get(columnIndex).getDataType());
+ }
+
+ @Override
+ public boolean isNull(int columnIndex) throws IOException {
+ return childrenColumns.get(columnIndex).isNull(index);
+ }
+
+ @Override
+ public int size() {
+ return childrenColumns.size();
+ }
+ };
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/TableUDFUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/TableUDFUtils.java
index bc2d42507a9..06bfa55feba 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/TableUDFUtils.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/utils/TableUDFUtils.java
@@ -25,30 +25,27 @@ import org.apache.iotdb.udf.api.relational.ScalarFunction;
import org.apache.iotdb.udf.api.relational.TableFunction;
public class TableUDFUtils {
- public static boolean isScalarFunction(String functionName) {
+ public static ScalarFunction tryGetScalarFunction(String functionName) {
try {
- UDFManagementService.getInstance().reflect(functionName, ScalarFunction.class);
- return true;
+ return UDFManagementService.getInstance().reflect(functionName, ScalarFunction.class);
} catch (Throwable e) {
- return false;
+ return null;
}
}
- public static boolean isTableFunction(String functionName) {
+ public static TableFunction tryGetTableFunction(String functionName) {
try {
- UDFManagementService.getInstance().reflect(functionName, TableFunction.class);
- return true;
+ return UDFManagementService.getInstance().reflect(functionName, TableFunction.class);
} catch (Throwable e) {
- return false;
+ return null;
}
}
- public static boolean isAggregateFunction(String functionName) {
+ public static AggregateFunction tryGetAggregateFunction(String functionName) {
try {
- UDFManagementService.getInstance().reflect(functionName, AggregateFunction.class);
- return true;
+ return UDFManagementService.getInstance().reflect(functionName, AggregateFunction.class);
} catch (Throwable e) {
- return false;
+ return null;
}
}
}