This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch DefaultOrderByTime
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/DefaultOrderByTime by this
push:
new b66e770e4e0 temp save
b66e770e4e0 is described below
commit b66e770e4e08962b736273e69aaf930d9648b5b1
Author: JackieTien97 <[email protected]>
AuthorDate: Thu May 15 17:18:35 2025 +0800
temp save
---
.../relational/analyzer/StatementAnalyzer.java | 6 ++
.../function/tvf/ForecastTableFunction.java | 56 ++++++++++++++++---
.../plan/relational/metadata/Metadata.java | 6 ++
.../relational/metadata/TableMetadataImpl.java | 9 +++
.../plan/relational/analyzer/TSBSMetadata.java | 6 ++
.../relational/analyzer/TableFunctionTest.java | 65 ++++++++++++++++++++++
.../plan/relational/analyzer/TestMetadata.java | 6 ++
7 files changed, 147 insertions(+), 7 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
index ef93fe1f5d8..1c15cf4c4a3 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
@@ -4061,6 +4061,12 @@ public class StatementAnalyzer {
public Scope visitTableFunctionInvocation(TableFunctionInvocation node,
Optional<Scope> scope) {
String functionName = node.getName().toString();
TableFunction function = metadata.getTableFunction(functionName);
+
+ // set model fetcher for ForecastTableFunction
+ if (function instanceof ForecastTableFunction) {
+ ((ForecastTableFunction)
function).setModelFetcher(metadata.getModelFetcher());
+ }
+
Node errorLocation = node;
if (!node.getArguments().isEmpty()) {
errorLocation = node.getArguments().get(0);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/ForecastTableFunction.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/ForecastTableFunction.java
index 534b0995da7..dbe91526bd8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/ForecastTableFunction.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/ForecastTableFunction.java
@@ -27,7 +27,6 @@ import
org.apache.iotdb.commons.client.ainode.AINodeClientManager;
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.plan.analyze.IModelFetcher;
-import org.apache.iotdb.db.queryengine.plan.analyze.ModelFetcher;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.model.ModelInferenceDescriptor;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.udf.api.relational.TableFunction;
@@ -65,6 +64,7 @@ import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
@@ -73,7 +73,7 @@ import static
org.apache.iotdb.rpc.TSStatusCode.CAN_NOT_CONNECT_AINODE;
public class ForecastTableFunction implements TableFunction {
- private static class ForecastTableFunctionHandle implements
TableFunctionHandle {
+ public static class ForecastTableFunctionHandle implements
TableFunctionHandle {
TEndPoint targetAINode;
String modelId;
int maxInputLength;
@@ -152,9 +152,41 @@ public class ForecastTableFunction implements
TableFunction {
types.add(Type.valueOf(ReadWriteIOUtils.readByte(buffer)));
}
}
- }
- private static final IModelFetcher MODEL_FETCHER =
ModelFetcher.getInstance();
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ForecastTableFunctionHandle that = (ForecastTableFunctionHandle) o;
+ return maxInputLength == that.maxInputLength
+ && outputLength == that.outputLength
+ && outputStartTime == that.outputStartTime
+ && outputInterval == that.outputInterval
+ && keepInput == that.keepInput
+ && Objects.equals(targetAINode, that.targetAINode)
+ && Objects.equals(modelId, that.modelId)
+ && Objects.equals(options, that.options)
+ && Objects.equals(types, that.types);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ targetAINode,
+ modelId,
+ maxInputLength,
+ outputLength,
+ outputStartTime,
+ outputInterval,
+ keepInput,
+ options,
+ types);
+ }
+ }
private static final String INPUT_PARAMETER_NAME = "INPUT";
private static final String MODEL_ID_PARAMETER_NAME = "MODEL_ID";
@@ -163,9 +195,9 @@ public class ForecastTableFunction implements TableFunction
{
private static final String PREDICATED_COLUMNS_PARAMETER_NAME =
"PREDICATED_COLUMNS";
private static final String DEFAULT_PREDICATED_COLUMNS = "";
private static final String OUTPUT_START_TIME = "OUTPUT_START_TIME";
- private static final long DEFAULT_OUTPUT_START_TIME = Long.MIN_VALUE;
+ public static final long DEFAULT_OUTPUT_START_TIME = Long.MIN_VALUE;
private static final String OUTPUT_INTERVAL = "OUTPUT_INTERVAL";
- private static final long DEFAULT_OUTPUT_INTERVAL = 0L;
+ public static final long DEFAULT_OUTPUT_INTERVAL = 0L;
public static final String TIMECOL_PARAMETER_NAME = "TIMECOL";
private static final String DEFAULT_TIME_COL = "time";
private static final String KEEP_INPUT_PARAMETER_NAME = "KEEP_INPUT";
@@ -185,6 +217,16 @@ public class ForecastTableFunction implements
TableFunction {
ALLOWED_INPUT_TYPES.add(Type.DOUBLE);
}
+ // need to set before analyze method is called
+ // should only be used in fe scope, never be used in
TableFunctionProcessorProvider
+ // The reason we don't directly set modelFetcher=ModelFetcher.getInstance()
is that we need to
+ // mock IModelFetcher in UT
+ private IModelFetcher modelFetcher = null;
+
+ public void setModelFetcher(IModelFetcher modelFetcher) {
+ this.modelFetcher = modelFetcher;
+ }
+
@Override
public List<ParameterSpecification> getArgumentsSpecifications() {
return Arrays.asList(
@@ -372,7 +414,7 @@ public class ForecastTableFunction implements TableFunction
{
}
private ModelInferenceDescriptor getModelInfo(String modelId) {
- return MODEL_FETCHER.fetchModel(modelId);
+ return modelFetcher.fetchModel(modelId);
}
// only allow for INT32, INT64, FLOAT, DOUBLE
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java
index 443fbd6e3f0..4d4c160cd50 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java
@@ -26,6 +26,7 @@ import
org.apache.iotdb.db.exception.load.LoadAnalyzeTableColumnDisorderExceptio
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
+import org.apache.iotdb.db.queryengine.plan.analyze.IModelFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.relational.function.OperatorType;
import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl;
@@ -194,4 +195,9 @@ public interface Metadata {
final String database, final List<DataPartitionQueryParam>
sgNameToQueryParamsMap);
TableFunction getTableFunction(final String functionName);
+
+ /**
+ * @return ModelFetcher
+ */
+ IModelFetcher getModelFetcher();
}
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 742f980e2e8..73703c8d4e2 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
@@ -34,7 +34,9 @@ import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
+import org.apache.iotdb.db.queryengine.plan.analyze.IModelFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import org.apache.iotdb.db.queryengine.plan.analyze.ModelFetcher;
import org.apache.iotdb.db.queryengine.plan.relational.function.OperatorType;
import
org.apache.iotdb.db.queryengine.plan.relational.function.TableBuiltinTableFunction;
import
org.apache.iotdb.db.queryengine.plan.relational.function.arithmetic.AdditionResolver;
@@ -95,6 +97,8 @@ public class TableMetadataImpl implements Metadata {
private final DataNodeTableCache tableCache =
DataNodeTableCache.getInstance();
+ private final IModelFetcher modelFetcher = ModelFetcher.getInstance();
+
@Override
public boolean tableExists(final QualifiedObjectName name) {
return tableCache.getTable(name.getDatabaseName(), name.getObjectName())
!= null;
@@ -841,6 +845,11 @@ public class TableMetadataImpl implements Metadata {
}
}
+ @Override
+ public IModelFetcher getModelFetcher() {
+ return modelFetcher;
+ }
+
public static boolean isTwoNumericType(List<? extends Type> argumentTypes) {
return argumentTypes.size() == 2
&& isNumericType(argumentTypes.get(0))
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java
index 189f86854c1..159ab2486a9 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java
@@ -28,6 +28,7 @@ import
org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
import org.apache.iotdb.commons.udf.builtin.BuiltinAggregationFunction;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
+import org.apache.iotdb.db.queryengine.plan.analyze.IModelFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.relational.function.OperatorType;
import
org.apache.iotdb.db.queryengine.plan.relational.metadata.AlignedDeviceEntry;
@@ -388,6 +389,11 @@ public class TSBSMetadata implements Metadata {
return null;
}
+ @Override
+ public IModelFetcher getModelFetcher() {
+ return null;
+ }
+
private static final DataPartition DATA_PARTITION =
MockTSBSDataPartition.constructDataPartition();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java
index 3aa3d85fa80..a1fddfe21d5 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TableFunctionTest.java
@@ -19,7 +19,9 @@
package org.apache.iotdb.db.queryengine.plan.relational.analyzer;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
+import
org.apache.iotdb.db.queryengine.plan.relational.function.tvf.ForecastTableFunction;
import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.TableFunctionProcessorMatcher;
@@ -32,8 +34,11 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import org.junit.Test;
+import java.util.Collections;
import java.util.function.Consumer;
+import static
org.apache.iotdb.db.queryengine.plan.relational.function.tvf.ForecastTableFunction.DEFAULT_OUTPUT_INTERVAL;
+import static
org.apache.iotdb.db.queryengine.plan.relational.function.tvf.ForecastTableFunction.DEFAULT_OUTPUT_START_TIME;
import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanAssert.assertPlan;
import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.aggregation;
import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.aggregationFunction;
@@ -48,6 +53,7 @@ import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions
import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.sort;
import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.tableFunctionProcessor;
import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.tableScan;
+import static org.apache.iotdb.udf.api.type.Type.DOUBLE;
public class TableFunctionTest {
@@ -332,4 +338,63 @@ public class TableFunctionTest {
deserialized.deserialize(serialized);
assert mapTableFunctionHandle.equals(deserialized);
}
+
+ @Test
+ public void testForecastFunction() {
+ // default order by time asc
+ PlanTester planTester = new PlanTester();
+
+ String sql =
+ "SELECT * FROM FORECAST("
+ + "input => (SELECT time,s3 FROM table1 WHERE tag1='shanghai' AND
tag2='A3' AND tag3='YY' ORDER BY time DESC LIMIT 1440), "
+ + "model_id => 'timer_xl'";
+ LogicalQueryPlan logicalQueryPlan = planTester.createPlan(sql);
+ PlanMatchPattern tableScan =
+ tableScan(
+ "testdb.table1",
+ ImmutableList.of("time", "tag1", "tag2", "tag3", "s3"),
+ ImmutableSet.of("time", "tag1", "tag2", "tag3", "s3"));
+ Consumer<TableFunctionProcessorMatcher.Builder> tableFunctionMatcher =
+ builder ->
+ builder
+ .name("forecast")
+ .properOutputs("time", "s3")
+ .requiredSymbols("time", "s3")
+ .handle(
+ new ForecastTableFunction.ForecastTableFunctionHandle(
+ false,
+ 1440,
+ "timer_xl",
+ Collections.emptyMap(),
+ 96,
+ DEFAULT_OUTPUT_START_TIME,
+ DEFAULT_OUTPUT_INTERVAL,
+ new TEndPoint("127.0.0.1", 10810),
+ Collections.singletonList(DOUBLE)));
+ // Verify full LogicalPlan
+ // Output - TableFunctionProcessor - TableScan
+ assertPlan(logicalQueryPlan,
anyTree(tableFunctionProcessor(tableFunctionMatcher, tableScan)));
+ // Verify DistributionPlan
+
+ /*
+ * └──OutputNode
+ * └──CollectNode
+ * ├──ExchangeNode
+ * │ └──TableFunctionProcessor
+ * │ └──TableScan
+ * ├──ExchangeNode
+ * │ └──TableFunctionProcessor
+ * │ └──TableScan
+ * └──ExchangeNode
+ * └──TableFunctionProcessor
+ * └──TableScan
+ */
+ assertPlan(planTester.getFragmentPlan(0), output(collect(exchange(),
exchange(), exchange())));
+ assertPlan(
+ planTester.getFragmentPlan(1),
tableFunctionProcessor(tableFunctionMatcher, tableScan));
+ assertPlan(
+ planTester.getFragmentPlan(2),
tableFunctionProcessor(tableFunctionMatcher, tableScan));
+ assertPlan(
+ planTester.getFragmentPlan(3),
tableFunctionProcessor(tableFunctionMatcher, tableScan));
+ }
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMetadata.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMetadata.java
index 3292a5077cd..e56727974a2 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMetadata.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMetadata.java
@@ -30,6 +30,7 @@ import
org.apache.iotdb.commons.udf.builtin.BuiltinAggregationFunction;
import org.apache.iotdb.commons.udf.builtin.relational.tvf.HOPTableFunction;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
+import org.apache.iotdb.db.queryengine.plan.analyze.IModelFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.function.Exclude;
import org.apache.iotdb.db.queryengine.plan.function.Repeat;
@@ -491,6 +492,11 @@ public class TestMetadata implements Metadata {
}
}
+ @Override
+ public IModelFetcher getModelFetcher() {
+ return null;
+ }
+
private static final DataPartition TABLE_DATA_PARTITION =
MockTableModelDataPartition.constructDataPartition(DB1);
