This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 0280aa7b51a Implement FFT table function for table model
0280aa7b51a is described below
commit 0280aa7b51a0c3c34401edd19165ee26f5f7358c
Author: DaZuiZui <[email protected]>
AuthorDate: Wed Aug 26 15:46:11 2026 +0800
Implement FFT table function for table model
---
.../it/db/it/IoTDBFFTTableFunctionIT.java | 293 ++++++++
.../relational/analyzer/StatementAnalyzer.java | 65 +-
.../plan/relational/planner/RelationPlanner.java | 3 +
.../distribute/TableDistributedPlanGenerator.java | 23 +-
.../relational/analyzer/TableFunctionTest.java | 174 +++++
.../apache/iotdb/commons/i18n/QueryMessages.java | 39 +
.../apache/iotdb/commons/i18n/QueryMessages.java | 36 +
.../function/TableBuiltinTableFunction.java | 4 +
.../builtin/relational/tvf/FFTTableFunction.java | 804 +++++++++++++++++++++
.../builtin/relational/tvf/fft/DoubleFFT_1D.java | 159 ++++
.../builtin/relational/tvf/fft/FloatFFT_1D.java | 160 ++++
.../relational/tvf/FFTTableFunctionTest.java | 518 +++++++++++++
.../udf/builtin/relational/tvf/fft/FFT1DTest.java | 109 +++
13 files changed, 2381 insertions(+), 6 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBFFTTableFunctionIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBFFTTableFunctionIT.java
new file mode 100644
index 00000000000..874f5bf9076
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBFFTTableFunctionIT.java
@@ -0,0 +1,293 @@
+/*
+ * 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.relational.it.db.it;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableClusterIT;
+import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.tableAssertTestFail;
+import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+public class IoTDBFFTTableFunctionIT {
+
+ private static final String DATABASE_NAME = "test_fft";
+ private static final double DELTA = 1e-9;
+ private static final String[] SQLS =
+ new String[] {
+ "CREATE DATABASE " + DATABASE_NAME,
+ "USE " + DATABASE_NAME,
+ "CREATE TABLE signal(device_id STRING TAG, temperature DOUBLE FIELD,
speed INT32 FIELD, note STRING FIELD)",
+ "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES
(0, 'd1', 1.0, 1, 'ok')",
+ "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES
(1000, 'd1', 0.0, 2, 'ok')",
+ "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES
(2000, 'd1', 0.0, 3, 'ok')",
+ "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES
(3000, 'd1', 0.0, 4, 'ok')",
+ "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES
(0, 'd2', 2.0, 4, 'ok')",
+ "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES
(1000, 'd2', 0.0, 3, 'ok')",
+ "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES
(2000, 'd2', 0.0, 2, 'ok')",
+ "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES
(3000, 'd2', 0.0, 1, 'ok')",
+ "CREATE TABLE single_row(device_id STRING TAG, value DOUBLE FIELD)",
+ "INSERT INTO single_row(time, device_id, value) VALUES (0, 'd1', 1.0)",
+ "CREATE TABLE no_numeric(device_id STRING TAG, note STRING FIELD)",
+ "INSERT INTO no_numeric(time, device_id, note) VALUES (0, 'd1', 'ok')",
+ "CREATE TABLE with_null(device_id STRING TAG, value DOUBLE FIELD)",
+ "INSERT INTO with_null(time, device_id, value) VALUES (0, 'd1', 1.0)",
+ "INSERT INTO with_null(time, device_id, value) VALUES (1000, 'd1',
null)",
+ "CREATE TABLE irregular(device_id STRING TAG, value DOUBLE FIELD)",
+ "INSERT INTO irregular(time, device_id, value) VALUES (0, 'd1', 1.0)",
+ "INSERT INTO irregular(time, device_id, value) VALUES (1000, 'd1',
2.0)",
+ "INSERT INTO irregular(time, device_id, value) VALUES (2500, 'd1',
3.0)",
+ "CREATE TABLE custom_time_signal(event_time TIMESTAMP TIME, value
DOUBLE FIELD)",
+ "INSERT INTO custom_time_signal(event_time, value) VALUES (0, 1.0)",
+ "INSERT INTO custom_time_signal(event_time, value) VALUES (1000, 0.0)",
+ "INSERT INTO custom_time_signal(event_time, value) VALUES (2000, 0.0)",
+ "INSERT INTO custom_time_signal(event_time, value) VALUES (3000, 0.0)",
+ "FLUSH"
+ };
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().initClusterEnvironment();
+ insertData();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ private static void insertData() {
+ String currentSql = null;
+ try (Connection connection = EnvFactory.getEnv().getTableConnection();
+ Statement statement = connection.createStatement()) {
+ for (String sql : SQLS) {
+ currentSql = sql;
+ statement.execute(sql);
+ }
+ } catch (Exception e) {
+ throw new AssertionError("insertData failed while executing [" +
currentSql + "].", e);
+ }
+ }
+
+ @Test
+ public void testFFTWithPartitionAndMultipleColumns() {
+ String[] expectedHeader =
+ new String[] {
+ "device_id",
+ "frequency_index",
+ "frequency",
+ "temperature_real",
+ "temperature_imag",
+ "speed_real",
+ "speed_imag"
+ };
+ assertFftRows(
+ "SELECT * FROM FFT(DATA => signal PARTITION BY device_id ORDER BY
time, "
+ + "SAMPLE_INTERVAL => 1s, N => 4) ORDER BY device_id,
frequency_index",
+ expectedHeader,
+ new Object[][] {
+ {"d1", 0L, 0.0, 1.0, 0.0, 10.0, 0.0},
+ {"d1", 1L, 0.25, 1.0, 0.0, -2.0, 2.0},
+ {"d1", 2L, -0.5, 1.0, 0.0, -2.0, 0.0},
+ {"d1", 3L, -0.25, 1.0, 0.0, -2.0, -2.0},
+ {"d2", 0L, 0.0, 2.0, 0.0, 10.0, 0.0},
+ {"d2", 1L, 0.25, 2.0, 0.0, 2.0, -2.0},
+ {"d2", 2L, -0.5, 2.0, 0.0, 2.0, 0.0},
+ {"d2", 3L, -0.25, 2.0, 0.0, 2.0, 2.0}
+ });
+ }
+
+ @Test
+ public void testFFTDefaultNAndInferredSampleInterval() {
+ String[] expectedHeader =
+ new String[] {"frequency_index", "frequency", "temperature_real",
"temperature_imag"};
+ String[] retArray =
+ new String[] {"0,0.0,1.0,0.0,", "1,0.25,1.0,0.0,", "2,-0.5,1.0,0.0,",
"3,-0.25,1.0,0.0,"};
+
+ tableResultSetEqualTest(
+ "SELECT frequency_index, frequency, temperature_real, temperature_imag
"
+ + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE
device_id='d1') "
+ + "ORDER BY time) ORDER BY frequency_index",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testFFTWithSpecifiedTimeColumn() {
+ tableResultSetEqualTest(
+ "SELECT frequency_index, frequency, value_real, value_imag "
+ + "FROM FFT(DATA => custom_time_signal ORDER BY event_time, "
+ + "TIMECOL => 'event_time', SAMPLE_INTERVAL => 1s, N => 4) "
+ + "ORDER BY frequency_index",
+ new String[] {"frequency_index", "frequency", "value_real",
"value_imag"},
+ new String[] {"0,0.0,1.0,0.0,", "1,0.25,1.0,0.0,", "2,-0.5,1.0,0.0,",
"3,-0.25,1.0,0.0,"},
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testFFTTruncateAndZeroPad() {
+ tableResultSetEqualTest(
+ "SELECT frequency_index, frequency, speed_real, speed_imag "
+ + "FROM FFT(DATA => (SELECT time, speed FROM signal WHERE
device_id='d1') "
+ + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 2) ORDER BY
frequency_index",
+ new String[] {"frequency_index", "frequency", "speed_real",
"speed_imag"},
+ new String[] {"0,0.0,3.0,0.0,", "1,-0.5,-1.0,0.0,"},
+ DATABASE_NAME);
+
+ tableResultSetEqualTest(
+ "SELECT frequency_index, frequency, temperature_real, temperature_imag
"
+ + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE
device_id='d1') "
+ + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 8) ORDER BY
frequency_index",
+ new String[] {"frequency_index", "frequency", "temperature_real",
"temperature_imag"},
+ new String[] {
+ "0,0.0,1.0,0.0,",
+ "1,0.125,1.0,0.0,",
+ "2,0.25,1.0,0.0,",
+ "3,0.375,1.0,0.0,",
+ "4,-0.5,1.0,0.0,",
+ "5,-0.375,1.0,0.0,",
+ "6,-0.25,1.0,0.0,",
+ "7,-0.125,1.0,0.0,"
+ },
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testFFTNorm() {
+ tableResultSetEqualTest(
+ "SELECT frequency_index, temperature_real, temperature_imag "
+ + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE
device_id='d1') "
+ + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 4, NORM =>
'forward') "
+ + "ORDER BY frequency_index",
+ new String[] {"frequency_index", "temperature_real",
"temperature_imag"},
+ new String[] {"0,0.25,0.0,", "1,0.25,0.0,", "2,0.25,0.0,",
"3,0.25,0.0,"},
+ DATABASE_NAME);
+
+ tableResultSetEqualTest(
+ "SELECT frequency_index, temperature_real, temperature_imag "
+ + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE
device_id='d1') "
+ + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 4, NORM => 'ortho') "
+ + "ORDER BY frequency_index",
+ new String[] {"frequency_index", "temperature_real",
"temperature_imag"},
+ new String[] {"0,0.5,0.0,", "1,0.5,0.0,", "2,0.5,0.0,", "3,0.5,0.0,"},
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testFFTIrregularTimeUsesInferredOrExplicitSampleInterval() {
+ String[] expectedHeader =
+ new String[] {"device_id", "frequency_index", "frequency",
"value_real", "value_imag"};
+
+ assertFftRows(
+ "SELECT * FROM FFT(DATA => irregular PARTITION BY device_id ORDER BY
time) "
+ + "ORDER BY frequency_index",
+ expectedHeader,
+ new Object[][] {
+ {"d1", 0L, 0.0, 6.0, 0.0},
+ {"d1", 1L, 0.26666666666666666, -1.5, 0.8660254037844386},
+ {"d1", 2L, -0.26666666666666666, -1.5, -0.8660254037844386}
+ });
+
+ assertFftRows(
+ "SELECT * FROM FFT(DATA => irregular PARTITION BY device_id ORDER BY
time, "
+ + "SAMPLE_INTERVAL => 1s) ORDER BY frequency_index",
+ expectedHeader,
+ new Object[][] {
+ {"d1", 0L, 0.0, 6.0, 0.0},
+ {"d1", 1L, 0.3333333333333333, -1.5, 0.8660254037844386},
+ {"d1", 2L, -0.3333333333333333, -1.5, -0.8660254037844386}
+ });
+ }
+
+ @Test
+ public void testFFTFailures() {
+ tableAssertTestFail(
+ "SELECT * FROM FFT(DATA => signal PARTITION BY device_id,
SAMPLE_INTERVAL => 1s)",
+ "701: Table argument with set semantics requires an ORDER BY clause.",
+ DATABASE_NAME);
+ tableAssertTestFail(
+ "SELECT * FROM FFT(DATA => single_row PARTITION BY device_id ORDER BY
time)",
+ "701: FFT requires at least two rows to infer SAMPLE_INTERVAL.",
+ DATABASE_NAME);
+ tableAssertTestFail(
+ "SELECT * FROM FFT(DATA => no_numeric PARTITION BY device_id ORDER BY
time, SAMPLE_INTERVAL => 1s)",
+ "701: No numeric columns found for FFT calculation.",
+ DATABASE_NAME);
+ tableAssertTestFail(
+ "SELECT * FROM FFT(DATA => with_null PARTITION BY device_id ORDER BY
time, SAMPLE_INTERVAL => 1s)",
+ "701: FFT does not support null values in column [value].",
+ DATABASE_NAME);
+ tableAssertTestFail(
+ "SELECT * FROM FFT(DATA => signal PARTITION BY device_id ORDER BY
time, N => 65537)",
+ "701: FFT transform length N must not exceed 65536.",
+ DATABASE_NAME);
+ }
+
+ private static void assertFftRows(String sql, String[] expectedHeader,
Object[][] expectedRows) {
+ try (Connection connection = EnvFactory.getEnv().getTableConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE " + DATABASE_NAME);
+ try (ResultSet resultSet = statement.executeQuery(sql)) {
+ assertHeader(resultSet.getMetaData(), expectedHeader);
+
+ int rowIndex = 0;
+ while (resultSet.next()) {
+ Object[] expectedRow = expectedRows[rowIndex];
+ assertEquals(expectedRow[0], resultSet.getString(1));
+ assertEquals(expectedRow[1], resultSet.getLong(2));
+ for (int columnIndex = 2; columnIndex < expectedRow.length;
columnIndex++) {
+ assertEquals(
+ (double) expectedRow[columnIndex],
resultSet.getDouble(columnIndex + 1), DELTA);
+ }
+ rowIndex++;
+ }
+ assertEquals(expectedRows.length, rowIndex);
+ }
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+ }
+
+ private static void assertHeader(ResultSetMetaData metaData, String[]
expectedHeader)
+ throws SQLException {
+ assertEquals(expectedHeader.length, metaData.getColumnCount());
+ for (int i = 1; i <= metaData.getColumnCount(); i++) {
+ assertEquals(expectedHeader[i - 1], metaData.getColumnName(i));
+ }
+ }
+}
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 741116c844c..9a54519953d 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
@@ -121,6 +121,7 @@ import
org.apache.iotdb.commons.queryengine.utils.cte.CteDataStore;
import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema;
+import org.apache.iotdb.commons.udf.builtin.relational.tvf.FFTTableFunction;
import org.apache.iotdb.commons.udf.builtin.relational.tvf.M4TableFunction;
import org.apache.iotdb.commons.udf.utils.UDFDataTypeTransformer;
import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
@@ -5775,7 +5776,7 @@ public class StatementAnalyzer {
Scope argumentScope = analysis.getScope(argument.getRelation());
if (argument.isPassThroughColumns()) {
argumentScope.getRelationType().getAllFields().forEach(fields::add);
- } else if
(!TableBuiltinTableFunction.M4.getFunctionName().equalsIgnoreCase(functionName)
+ } else if (!isPartitionColumnsProvidedByProperSchema(functionName)
&& argument.getPartitionBy().isPresent()) {
argument.getPartitionBy().get().stream()
.map(expression -> validateAndGetInputField(expression,
argumentScope))
@@ -5915,9 +5916,71 @@ public class StatementAnalyzer {
}
}
tryAppendM4ModeArgument(functionName, arguments,
parameterSpecifications, passedArguments);
+ tryAppendFFTInternalArguments(
+ functionName, arguments, parameterSpecifications, passedArguments);
return new ArgumentsAnalysis(passedArguments.buildOrThrow(),
tableArgumentAnalyses.build());
}
+ private boolean isPartitionColumnsProvidedByProperSchema(String
functionName) {
+ return
TableBuiltinTableFunction.M4.getFunctionName().equalsIgnoreCase(functionName)
+ ||
TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName);
+ }
+
+ private void tryAppendFFTInternalArguments(
+ String functionName,
+ List<TableFunctionArgument> arguments,
+ List<ParameterSpecification> parameterSpecifications,
+ ImmutableMap.Builder<String, Argument> passedArguments) {
+ if
(!TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName))
{
+ return;
+ }
+
+ Optional<TableFunctionArgument> sampleIntervalArgument =
+ findOptionalTableFunctionArgument(
+ arguments, parameterSpecifications,
FFTTableFunction.SAMPLE_INTERVAL_PARAMETER_NAME);
+ if (sampleIntervalArgument.isPresent()
+ && !(sampleIntervalArgument.get().getValue() instanceof
TimeDurationLiteral)) {
+ throw new
SemanticException(QueryMessages.FFT_SAMPLE_INTERVAL_MUST_BE_DURATION_LITERAL);
+ }
+
+ Optional<TableFunctionArgument> nArgument =
+ findOptionalTableFunctionArgument(
+ arguments, parameterSpecifications,
FFTTableFunction.N_PARAMETER_NAME);
+ if (nArgument.isPresent() && nArgument.get().getValue() instanceof
TimeDurationLiteral) {
+ throw new
SemanticException(QueryMessages.FFT_N_MUST_BE_POSITIVE_INTEGER);
+ }
+
+ validateFFTOrderBySortOrder(arguments, parameterSpecifications);
+ passedArguments.put(
+ FFTTableFunction.SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME,
+ new ScalarArgument(
+ org.apache.iotdb.udf.api.type.Type.BOOLEAN,
sampleIntervalArgument.isPresent()));
+ }
+
+ private void validateFFTOrderBySortOrder(
+ List<TableFunctionArgument> arguments,
+ List<ParameterSpecification> parameterSpecifications) {
+ Optional<TableFunctionArgument> dataArgument =
+ findOptionalTableFunctionArgument(
+ arguments, parameterSpecifications,
FFTTableFunction.DATA_PARAMETER_NAME);
+ if (!dataArgument.isPresent()
+ || !(dataArgument.get().getValue() instanceof
TableFunctionTableArgument)) {
+ return;
+ }
+
+ Optional<OrderBy> orderBy =
+ ((TableFunctionTableArgument)
dataArgument.get().getValue()).getOrderBy();
+ if (!orderBy.isPresent()) {
+ return;
+ }
+
+ for (SortItem sortItem : orderBy.get().getSortItems()) {
+ if (sortItem.getOrdering() != SortItem.Ordering.ASCENDING) {
+ throw new
SemanticException(QueryMessages.FFT_ORDER_BY_MUST_SORT_ASCENDING);
+ }
+ }
+ }
+
private void tryAppendM4ModeArgument(
String functionName,
List<TableFunctionArgument> arguments,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
index 117d5db4e5c..7fc367b3785 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
@@ -1575,6 +1575,9 @@ public class RelationPlanner implements
AstVisitor<RelationPlan, Void> {
} else if (!TableBuiltinTableFunction.M4
.getFunctionName()
.equalsIgnoreCase(functionAnalysis.getFunctionName())
+ && !TableBuiltinTableFunction.FFT
+ .getFunctionName()
+ .equalsIgnoreCase(functionAnalysis.getFunctionName())
&& tableArgument.getPartitionBy().isPresent()) {
tableArgument.getPartitionBy().get().stream()
// the original symbols for partitioning columns, not coerced
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
index 25845758a08..60a9cd2cba2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
@@ -32,6 +32,7 @@ import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.process.SingleChildProcessNode;
import
org.apache.iotdb.commons.queryengine.plan.relational.function.BoundSignature;
import
org.apache.iotdb.commons.queryengine.plan.relational.function.FunctionId;
+import
org.apache.iotdb.commons.queryengine.plan.relational.function.TableBuiltinTableFunction;
import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema;
import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName;
import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.ResolvedFunction;
@@ -1906,22 +1907,34 @@ public class TableDistributedPlanGenerator
if (node.getChildren().isEmpty()) {
return Collections.singletonList(node);
}
- boolean canSplitPushDown = node.isRowSemantic() || (node.getChild()
instanceof GroupNode);
+ boolean canSplitPushDown = canSplitTableFunctionProcessor(node);
List<PlanNode> childrenNodes = node.getChild().accept(this, context);
if (childrenNodes.size() == 1) {
node.setChild(childrenNodes.get(0));
return Collections.singletonList(node);
} else if (!canSplitPushDown) {
- CollectNode collectNode =
- new CollectNode(queryId.genPlanNodeId(),
node.getChildren().get(0).getOutputSymbols());
- childrenNodes.forEach(collectNode::addChild);
- node.setChild(collectNode);
+ OrderingScheme childOrdering =
nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId());
+ node.setChild(mergeChildrenViaCollectOrMergeSort(childOrdering,
childrenNodes));
return Collections.singletonList(node);
} else {
return splitForEachChild(node, childrenNodes);
}
}
+ private boolean canSplitTableFunctionProcessor(TableFunctionProcessorNode
node) {
+ if (node.isRowSemantic()) {
+ return true;
+ }
+ if (!isPartitionedGroup(node.getChild())) {
+ return false;
+ }
+ return
!TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(node.getName());
+ }
+
+ private boolean isPartitionedGroup(PlanNode node) {
+ return node instanceof GroupNode && ((GroupNode)
node).getPartitionKeyCount() > 0;
+ }
+
private void buildRegionNodeMap(
AggregationTableScanNode originalAggTableScanNode,
List<List<TRegionReplicaSet>> regionReplicaSetsList,
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 9d09ff057b3..637113026f4 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
@@ -22,6 +22,7 @@ package
org.apache.iotdb.db.queryengine.plan.relational.analyzer;
import org.apache.iotdb.commons.exception.SemanticException;
import
org.apache.iotdb.commons.queryengine.plan.relational.function.tvf.ForecastTableFunction;
import
org.apache.iotdb.commons.queryengine.plan.relational.planner.node.JoinNode;
+import org.apache.iotdb.commons.udf.builtin.relational.tvf.FFTTableFunction;
import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern;
@@ -482,6 +483,179 @@ public class TableFunctionTest {
}
}
+ @Test
+ public void testFFTFunction() {
+ PlanTester planTester = new PlanTester();
+ String sql =
+ "SELECT * FROM FFT("
+ + "DATA => table1 PARTITION BY tag1 ORDER BY time, "
+ + "SAMPLE_INTERVAL => 1s, "
+ + "N => 4, "
+ + "NORM => 'ortho')";
+ LogicalQueryPlan logicalQueryPlan = planTester.createPlan(sql);
+ PlanMatchPattern tableScan =
+ tableScan(
+ "testdb.table1",
+ ImmutableMap.<String, String>builder()
+ .put("time", "time")
+ .put("tag1_0", "tag1")
+ .put("s1", "s1")
+ .put("s2", "s2")
+ .put("s3", "s3")
+ .buildOrThrow());
+
+ Consumer<TableFunctionProcessorMatcher.Builder> tableFunctionMatcher =
+ builder ->
+ builder
+ .name("fft")
+ .properOutputs(
+ "tag1",
+ "frequency_index",
+ "frequency",
+ "s1_real",
+ "s1_imag",
+ "s2_real",
+ "s2_imag",
+ "s3_real",
+ "s3_imag")
+ .requiredSymbols("time", "tag1_0", "s1", "s2", "s3")
+ .handle(
+ new MapTableFunctionHandle.Builder()
+
.addProperty(FFTTableFunction.SAMPLE_INTERVAL_PARAMETER_NAME, 1000L)
+ .addProperty(
+
FFTTableFunction.SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME, true)
+ .addProperty(FFTTableFunction.N_PARAMETER_NAME, 4L)
+ .addProperty(FFTTableFunction.NORM_PARAMETER_NAME,
"ortho")
+ .addProperty("__FFT_PARTITION_TYPES", "STRING")
+ .addProperty("__FFT_VALUE_TYPES", "INT64,INT64,DOUBLE")
+ .addProperty("__FFT_VALUE_NAMES", "czE=,czI=,czM=")
+ .build());
+
+ assertPlan(
+ logicalQueryPlan, anyTree(tableFunctionProcessor(tableFunctionMatcher,
sort(tableScan))));
+ assertPlan(
+ planTester.getFragmentPlan(0),
+ output(
+ tableFunctionProcessor(
+ tableFunctionMatcher, mergeSort(exchange(), exchange(),
exchange()))));
+ assertPlan(planTester.getFragmentPlan(1), sort(tableScan));
+ assertPlan(planTester.getFragmentPlan(2), sort(tableScan));
+ assertPlan(planTester.getFragmentPlan(3), sort(tableScan));
+ }
+
+ @Test
+ public void testFFTDefaultArguments() {
+ PlanTester planTester = new PlanTester();
+ String sql = "SELECT * FROM TABLE(FFT(DATA => TABLE(table1) ORDER BY
time))";
+ LogicalQueryPlan logicalQueryPlan = planTester.createPlan(sql);
+ PlanMatchPattern tableScan =
+ tableScan(
+ "testdb.table1",
+ ImmutableMap.<String, String>builder()
+ .put("time", "time")
+ .put("s1", "s1")
+ .put("s2", "s2")
+ .put("s3", "s3")
+ .buildOrThrow());
+
+ Consumer<TableFunctionProcessorMatcher.Builder> tableFunctionMatcher =
+ builder ->
+ builder
+ .name("fft")
+ .properOutputs(
+ "frequency_index",
+ "frequency",
+ "s1_real",
+ "s1_imag",
+ "s2_real",
+ "s2_imag",
+ "s3_real",
+ "s3_imag")
+ .requiredSymbols("time", "s1", "s2", "s3")
+ .handle(
+ new MapTableFunctionHandle.Builder()
+ .addProperty(
+ FFTTableFunction.SAMPLE_INTERVAL_PARAMETER_NAME,
Long.MIN_VALUE)
+ .addProperty(
+
FFTTableFunction.SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME, false)
+ .addProperty(FFTTableFunction.N_PARAMETER_NAME, -1L)
+ .addProperty(FFTTableFunction.NORM_PARAMETER_NAME,
"backward")
+ .addProperty("__FFT_PARTITION_TYPES", "")
+ .addProperty("__FFT_VALUE_TYPES", "INT64,INT64,DOUBLE")
+ .addProperty("__FFT_VALUE_NAMES", "czE=,czI=,czM=")
+ .build());
+
+ assertPlan(
+ logicalQueryPlan, anyTree(tableFunctionProcessor(tableFunctionMatcher,
sort(tableScan))));
+ assertPlan(
+ planTester.getFragmentPlan(0),
+ output(
+ tableFunctionProcessor(
+ tableFunctionMatcher, mergeSort(exchange(), exchange(),
exchange()))));
+ assertPlan(planTester.getFragmentPlan(1), sort(tableScan));
+ assertPlan(planTester.getFragmentPlan(2), sort(tableScan));
+ assertPlan(planTester.getFragmentPlan(3), sort(tableScan));
+ }
+
+ @Test
+ public void testFFTWithSpecifiedTimeColumn() {
+ PlanTester planTester = new PlanTester();
+ String sql =
+ "SELECT * FROM FFT("
+ + "DATA => (SELECT time AS event_time, tag1, s1 FROM table1) "
+ + "PARTITION BY tag1 ORDER BY event_time, "
+ + "TIMECOL => 'event_time', "
+ + "SAMPLE_INTERVAL => 1s, "
+ + "N => 4)";
+
+ planTester.createPlan(sql);
+ }
+
+ @Test
+ public void testFFTPositionalArgumentsKeepExistingOrder() {
+ PlanTester planTester = new PlanTester();
+ String sql = "SELECT * FROM TABLE(FFT(TABLE(table1) ORDER BY time, 1s, 4,
'ortho'))";
+
+ planTester.createPlan(sql);
+ }
+
+ @Test
+ public void testFFTRejectsInvalidArguments() {
+ assertAnalyzeFails(
+ "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1, SAMPLE_INTERVAL
=> 1ms)",
+ "Table argument with set semantics requires an ORDER BY clause.");
+ assertAnalyzeFails(
+ "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time
DESC, SAMPLE_INTERVAL => 1ms)",
+ "The ORDER BY clause of the DATA argument must sort the time column in
ascending order.");
+ assertAnalyzeFails(
+ "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY s1,
SAMPLE_INTERVAL => 1ms)",
+ "The ORDER BY clause of the DATA argument must contain exactly the
time column specified by the TIMECOL argument.");
+ assertAnalyzeFails(
+ "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time,
SAMPLE_INTERVAL => 1)",
+ "The SAMPLE_INTERVAL argument of FFT must be a duration literal.");
+ assertAnalyzeFails(
+ "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time, N
=> 0)",
+ "Invalid scalar argument N, should be a positive value");
+ assertAnalyzeFails(
+ "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time, N
=> 65537)",
+ "FFT transform length N must not exceed 65536.");
+ assertAnalyzeFails(
+ "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time,
NORM => 'bad')",
+ "Invalid NORM value for FFT. Supported values are backward, forward
and ortho.");
+ assertAnalyzeFails(
+ "SELECT * FROM FFT(DATA => (SELECT time, tag1 FROM table1) PARTITION
BY tag1 ORDER BY time)",
+ "No numeric columns found for FFT calculation.");
+ }
+
+ private void assertAnalyzeFails(String sql, String message) {
+ try {
+ analyzeSQL(sql, TEST_MATADATA, QUERY_CONTEXT);
+ fail();
+ } catch (SemanticException e) {
+ assertEquals(message, e.getMessage());
+ }
+ }
+
@Test
public void testM4TimeWindowMode() {
PlanTester planTester = new PlanTester();
diff --git
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/QueryMessages.java
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/QueryMessages.java
index fdee28c8cad..e84ebc6d9a9 100644
---
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/QueryMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/QueryMessages.java
@@ -554,4 +554,43 @@ public final class QueryMessages {
public static final String EXCEPTION_MISSING_REQUIRED_ARGUMENT_ARG_1308D0DC
= "Missing required argument: %s";
public static final String EXCEPTION_UNKNOWN_ARGUMENT_ARG_F40E9394 =
"Unknown argument: %s";
+ // ======================== FFT table function ========================
+
+ public static final String FFT_DATA_REQUIRES_ORDER_BY =
+ "Table argument with set semantics requires an ORDER BY clause.";
+ public static final String FFT_REQUIRES_NAMED_NUMERIC_COLUMNS =
+ "FFT requires named numeric input columns.";
+ public static final String FFT_NO_NUMERIC_COLUMNS =
+ "No numeric columns found for FFT calculation.";
+ public static final String FFT_ORDER_BY_MUST_CONTAIN_TIMECOL =
+ "The ORDER BY clause of the DATA argument must contain exactly the time
column specified by the TIMECOL argument.";
+ public static final String FFT_TRANSFORM_LENGTH_EXCEEDS_LIMIT =
+ "FFT transform length N must not exceed %d.";
+ public static final String FFT_SPECTRUM_BUFFER_TOO_LARGE =
+ "FFT spectrum buffer is too large. Reduce N or the number of numeric
columns.";
+ public static final String FFT_INVALID_NORM =
+ "Invalid NORM value for FFT. Supported values are backward, forward and
ortho.";
+ public static final String FFT_UNSUPPORTED_PARTITION_TYPE =
+ "Unsupported FFT partition type: %s";
+ public static final String FFT_UNSUPPORTED_VALUE_TYPE = "Unsupported FFT
value type: %s";
+ public static final String FFT_TIME_MUST_BE_STRICTLY_ASCENDING =
+ "The time column of FFT input must be strictly ascending within each
partition.";
+ public static final String FFT_NULL_VALUE_NOT_SUPPORTED =
+ "FFT does not support null values in column [%s].";
+ public static final String FFT_NEEDS_TWO_ROWS_FOR_INTERVAL =
+ "FFT requires at least two rows to infer SAMPLE_INTERVAL.";
+ public static final String FFT_SAMPLE_INTERVAL_MUST_BE_POSITIVE =
+ "FFT SAMPLE_INTERVAL must be positive.";
+ public static final String FFT_SAMPLE_INTERVAL_MUST_BE_DURATION_LITERAL =
+ "The SAMPLE_INTERVAL argument of FFT must be a duration literal.";
+ public static final String FFT_N_MUST_BE_POSITIVE_INTEGER =
+ "The N argument of FFT must be a positive integer.";
+ public static final String FFT_ORDER_BY_MUST_SORT_ASCENDING =
+ "The ORDER BY clause of the DATA argument must sort the time column in
ascending order.";
+ public static final String
EXCEPTION_FFT_LENGTH_MUST_BE_A_POSITIVE_INT_SIZED_VALUE_A000D3BB =
+ "FFT length must be a positive int-sized value.";
+ public static final String
+ EXCEPTION_INPUT_ARRAY_LENGTH_MUST_BE_AT_LEAST_2_FFT_LENGTH_31DF6A25 =
+ "Input array length must be at least 2 * FFT length.";
+
}
diff --git
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/QueryMessages.java
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/QueryMessages.java
index 6fa67c0a65d..03aab191920 100644
---
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/QueryMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/QueryMessages.java
@@ -554,4 +554,40 @@ public final class QueryMessages {
public static final String EXCEPTION_MISSING_REQUIRED_ARGUMENT_ARG_1308D0DC
= "缺少必要参数：%s";
public static final String EXCEPTION_UNKNOWN_ARGUMENT_ARG_F40E9394 =
"未知参数：%s";
+ // ======================== FFT table function ========================
+
+ public static final String FFT_DATA_REQUIRES_ORDER_BY =
+ "具有 set 语义的表参数需要 ORDER BY 子句。";
+ public static final String FFT_REQUIRES_NAMED_NUMERIC_COLUMNS =
+ "FFT 要求数值输入列必须有名称。";
+ public static final String FFT_NO_NUMERIC_COLUMNS = "未找到用于 FFT 计算的数值列。";
+ public static final String FFT_ORDER_BY_MUST_CONTAIN_TIMECOL =
+ "DATA 参数的 ORDER BY 子句必须恰好包含 TIMECOL 参数指定的时间列。";
+ public static final String FFT_TRANSFORM_LENGTH_EXCEEDS_LIMIT =
+ "FFT 变换长度 N 不能超过 %d。";
+ public static final String FFT_SPECTRUM_BUFFER_TOO_LARGE =
+ "FFT 频谱缓冲区过大。请减小 N 或数值列的数量。";
+ public static final String FFT_INVALID_NORM =
+ "FFT 的 NORM 值无效。支持 backward、forward 和 ortho。";
+ public static final String FFT_UNSUPPORTED_PARTITION_TYPE = "不支持的 FFT
分区类型：%s";
+ public static final String FFT_UNSUPPORTED_VALUE_TYPE = "不支持的 FFT 数值类型：%s";
+ public static final String FFT_TIME_MUST_BE_STRICTLY_ASCENDING =
+ "FFT 输入的时间列在每个分区内必须严格递增。";
+ public static final String FFT_NULL_VALUE_NOT_SUPPORTED =
+ "FFT 不支持列 [%s] 中的 null 值。";
+ public static final String FFT_NEEDS_TWO_ROWS_FOR_INTERVAL =
+ "FFT 至少需要两行数据才能推断 SAMPLE_INTERVAL。";
+ public static final String FFT_SAMPLE_INTERVAL_MUST_BE_POSITIVE =
+ "FFT 的 SAMPLE_INTERVAL 必须为正数。";
+ public static final String FFT_SAMPLE_INTERVAL_MUST_BE_DURATION_LITERAL =
+ "FFT 的 SAMPLE_INTERVAL 参数必须是 duration 字面量。";
+ public static final String FFT_N_MUST_BE_POSITIVE_INTEGER = "FFT 的 N
参数必须是正整数。";
+ public static final String FFT_ORDER_BY_MUST_SORT_ASCENDING =
+ "DATA 参数的 ORDER BY 子句必须按时间列升序排序。";
+ public static final String
EXCEPTION_FFT_LENGTH_MUST_BE_A_POSITIVE_INT_SIZED_VALUE_A000D3BB =
+ "FFT 长度必须是正数且不能超过 int 范围。";
+ public static final String
+ EXCEPTION_INPUT_ARRAY_LENGTH_MUST_BE_AT_LEAST_2_FFT_LENGTH_31DF6A25 =
+ "输入数组长度必须至少为 FFT 长度的 2 倍。";
+
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/function/TableBuiltinTableFunction.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/function/TableBuiltinTableFunction.java
index eb969a2c6ae..decaf8dc466 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/function/TableBuiltinTableFunction.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/function/TableBuiltinTableFunction.java
@@ -25,6 +25,7 @@ import
org.apache.iotdb.commons.queryengine.plan.relational.function.tvf.Forecas
import
org.apache.iotdb.commons.queryengine.plan.relational.function.tvf.PatternMatchTableFunction;
import
org.apache.iotdb.commons.udf.builtin.relational.tvf.CapacityTableFunction;
import
org.apache.iotdb.commons.udf.builtin.relational.tvf.CumulateTableFunction;
+import org.apache.iotdb.commons.udf.builtin.relational.tvf.FFTTableFunction;
import org.apache.iotdb.commons.udf.builtin.relational.tvf.HOPTableFunction;
import org.apache.iotdb.commons.udf.builtin.relational.tvf.M4TableFunction;
import
org.apache.iotdb.commons.udf.builtin.relational.tvf.SessionTableFunction;
@@ -45,6 +46,7 @@ public enum TableBuiltinTableFunction {
VARIATION("variation"),
CAPACITY("capacity"),
M4("m4"),
+ FFT("fft"),
FORECAST("forecast"),
PATTERN_MATCH("pattern_match"),
CLASSIFY("classify");
@@ -91,6 +93,8 @@ public enum TableBuiltinTableFunction {
return new CapacityTableFunction();
case "m4":
return new M4TableFunction();
+ case "fft":
+ return new FFTTableFunction();
case "forecast":
return new ForecastTableFunction();
case "classify":
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunction.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunction.java
new file mode 100644
index 00000000000..ebdd089c172
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunction.java
@@ -0,0 +1,804 @@
+/*
+ * 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.builtin.relational.tvf;
+
+import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.i18n.QueryMessages;
+import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils;
+import org.apache.iotdb.commons.udf.builtin.relational.tvf.fft.DoubleFFT_1D;
+import org.apache.iotdb.commons.udf.builtin.relational.tvf.fft.FloatFFT_1D;
+import org.apache.iotdb.udf.api.exception.UDFException;
+import org.apache.iotdb.udf.api.relational.TableFunction;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.relational.table.MapTableFunctionHandle;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionHandle;
+import
org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider;
+import org.apache.iotdb.udf.api.relational.table.argument.Argument;
+import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema;
+import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
+import org.apache.iotdb.udf.api.relational.table.argument.TableArgument;
+import
org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor;
+import
org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification;
+import
org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification;
+import
org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification;
+import org.apache.iotdb.udf.api.type.Type;
+
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.utils.Binary;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+import static
org.apache.iotdb.commons.udf.builtin.relational.tvf.WindowTVFUtils.findColumnIndex;
+import static
org.apache.iotdb.udf.api.relational.table.argument.ScalarArgumentChecker.POSITIVE_LONG_CHECKER;
+
+public class FFTTableFunction implements TableFunction {
+
+ public static final String DATA_PARAMETER_NAME = "DATA";
+ public static final String TIMECOL_PARAMETER_NAME = "TIMECOL";
+ public static final String SAMPLE_INTERVAL_PARAMETER_NAME =
"SAMPLE_INTERVAL";
+ public static final String N_PARAMETER_NAME = "N";
+ public static final String NORM_PARAMETER_NAME = "NORM";
+ public static final String SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME =
+ "__FFT_SAMPLE_INTERVAL_SPECIFIED";
+
+ private static final String DEFAULT_TIME_COLUMN_NAME = "time";
+ private static final String OUTPUT_FREQUENCY_INDEX_COLUMN =
"frequency_index";
+ private static final String OUTPUT_FREQUENCY_COLUMN = "frequency";
+ private static final String PARTITION_TYPES_PROPERTY =
"__FFT_PARTITION_TYPES";
+ private static final String VALUE_TYPES_PROPERTY = "__FFT_VALUE_TYPES";
+ private static final String VALUE_NAMES_PROPERTY = "__FFT_VALUE_NAMES";
+ private static final long UNSPECIFIED_SAMPLE_INTERVAL = Long.MIN_VALUE;
+ private static final long UNSPECIFIED_N = -1L;
+ private static final long MAX_TRANSFORM_LENGTH = 65_536L;
+ private static final long MAX_SPECTRUM_VALUES = 16_777_216L;
+ private static final String NORM_BACKWARD = "backward";
+ private static final String NORM_FORWARD = "forward";
+ private static final String NORM_ORTHO = "ortho";
+ private static final Set<Type> SUPPORTED_PARTITION_TYPES =
+ new HashSet<>(
+ Arrays.asList(
+ Type.BOOLEAN,
+ Type.INT32,
+ Type.INT64,
+ Type.FLOAT,
+ Type.DOUBLE,
+ Type.TEXT,
+ Type.TIMESTAMP,
+ Type.DATE,
+ Type.BLOB,
+ Type.STRING));
+ private static final Set<Type> SUPPORTED_VALUE_TYPES =
+ new HashSet<>(Arrays.asList(Type.INT32, Type.INT64, Type.FLOAT,
Type.DOUBLE));
+
+ @Override
+ public List<ParameterSpecification> getArgumentsSpecifications() {
+ return Arrays.asList(
+
TableParameterSpecification.builder().name(DATA_PARAMETER_NAME).setSemantics().build(),
+ ScalarParameterSpecification.builder()
+ .name(SAMPLE_INTERVAL_PARAMETER_NAME)
+ .type(Type.INT64)
+ .defaultValue(UNSPECIFIED_SAMPLE_INTERVAL)
+ .addChecker(POSITIVE_LONG_CHECKER)
+ .build(),
+ ScalarParameterSpecification.builder()
+ .name(N_PARAMETER_NAME)
+ .type(Type.INT64)
+ .defaultValue(UNSPECIFIED_N)
+ .addChecker(POSITIVE_LONG_CHECKER)
+ .build(),
+ ScalarParameterSpecification.builder()
+ .name(NORM_PARAMETER_NAME)
+ .type(Type.STRING)
+ .defaultValue(NORM_BACKWARD)
+ .build(),
+ ScalarParameterSpecification.builder()
+ .name(TIMECOL_PARAMETER_NAME)
+ .type(Type.STRING)
+ .defaultValue(DEFAULT_TIME_COLUMN_NAME)
+ .build());
+ }
+
+ @Override
+ public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws
UDFException {
+ TableArgument tableArgument = (TableArgument)
arguments.get(DATA_PARAMETER_NAME);
+ if (tableArgument.getOrderBy().isEmpty()) {
+ throw new SemanticException(QueryMessages.FFT_DATA_REQUIRES_ORDER_BY);
+ }
+
+ String timeColumn =
+ (String) ((ScalarArgument)
arguments.get(TIMECOL_PARAMETER_NAME)).getValue();
+ int timeColumnIndex =
+ findColumnIndex(tableArgument, timeColumn,
Collections.singleton(Type.TIMESTAMP));
+ validateOrderBy(tableArgument, timeColumn);
+
+ List<Integer> partitionIndexes = getPartitionIndexes(tableArgument);
+ Set<Integer> excludedIndexes = new HashSet<>(partitionIndexes);
+ excludedIndexes.add(timeColumnIndex);
+
+ List<Integer> valueIndexes = new ArrayList<>();
+ List<String> valueNames = new ArrayList<>();
+ List<Type> valueTypes = new ArrayList<>();
+ List<Type> partitionTypes = new ArrayList<>();
+ DescribedSchema.Builder schemaBuilder = new DescribedSchema.Builder();
+
+ for (int partitionIndex : partitionIndexes) {
+ Type type = tableArgument.getFieldTypes().get(partitionIndex);
+ partitionTypes.add(type);
+
schemaBuilder.addField(tableArgument.getFieldNames().get(partitionIndex).get(),
type);
+ }
+ schemaBuilder
+ .addField(OUTPUT_FREQUENCY_INDEX_COLUMN, Type.INT64)
+ .addField(OUTPUT_FREQUENCY_COLUMN, Type.DOUBLE);
+
+ for (int i = 0; i < tableArgument.getFieldTypes().size(); i++) {
+ if (excludedIndexes.contains(i)) {
+ continue;
+ }
+ Type type = tableArgument.getFieldTypes().get(i);
+ if (!SUPPORTED_VALUE_TYPES.contains(type)) {
+ continue;
+ }
+ String columnName =
+ tableArgument
+ .getFieldNames()
+ .get(i)
+ .orElseThrow(
+ () -> new
SemanticException(QueryMessages.FFT_REQUIRES_NAMED_NUMERIC_COLUMNS));
+ valueIndexes.add(i);
+ valueNames.add(columnName);
+ valueTypes.add(type);
+ schemaBuilder.addField(columnName + "_real", Type.DOUBLE);
+ schemaBuilder.addField(columnName + "_imag", Type.DOUBLE);
+ }
+
+ if (valueIndexes.isEmpty()) {
+ throw new SemanticException(QueryMessages.FFT_NO_NUMERIC_COLUMNS);
+ }
+
+ long transformLength = (long) ((ScalarArgument)
arguments.get(N_PARAMETER_NAME)).getValue();
+ validateTransformLength(transformLength, valueIndexes.size());
+ String norm =
+ ((String) ((ScalarArgument)
arguments.get(NORM_PARAMETER_NAME)).getValue())
+ .toLowerCase(Locale.ROOT);
+ validateNorm(norm);
+
+ MapTableFunctionHandle handle =
+ new MapTableFunctionHandle.Builder()
+ .addProperty(
+ SAMPLE_INTERVAL_PARAMETER_NAME,
+ ((ScalarArgument)
arguments.get(SAMPLE_INTERVAL_PARAMETER_NAME)).getValue())
+ .addProperty(
+ SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME,
+ (boolean)
+ ((ScalarArgument)
arguments.get(SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME))
+ .getValue())
+ .addProperty(N_PARAMETER_NAME, transformLength)
+ .addProperty(NORM_PARAMETER_NAME, norm)
+ .addProperty(PARTITION_TYPES_PROPERTY, joinTypes(partitionTypes))
+ .addProperty(VALUE_TYPES_PROPERTY, joinTypes(valueTypes))
+ .addProperty(VALUE_NAMES_PROPERTY, encodeStrings(valueNames))
+ .build();
+
+ List<Integer> requiredColumns = new ArrayList<>();
+ requiredColumns.add(timeColumnIndex);
+ requiredColumns.addAll(partitionIndexes);
+ requiredColumns.addAll(valueIndexes);
+
+ return TableFunctionAnalysis.builder()
+ .properColumnSchema(schemaBuilder.build())
+ .requireRecordSnapshot(false)
+ .requiredColumns(DATA_PARAMETER_NAME, requiredColumns)
+ .handle(handle)
+ .build();
+ }
+
+ @Override
+ public TableFunctionHandle createTableFunctionHandle() {
+ return new MapTableFunctionHandle();
+ }
+
+ @Override
+ public TableFunctionProcessorProvider getProcessorProvider(
+ TableFunctionHandle tableFunctionHandle) {
+ MapTableFunctionHandle handle = (MapTableFunctionHandle)
tableFunctionHandle;
+ boolean sampleIntervalSpecified =
+ (boolean) handle.getProperty(SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME);
+ long sampleInterval = (long)
handle.getProperty(SAMPLE_INTERVAL_PARAMETER_NAME);
+ long transformLength = (long) handle.getProperty(N_PARAMETER_NAME);
+ String norm = (String) handle.getProperty(NORM_PARAMETER_NAME);
+ Type[] partitionTypes = parseTypes((String)
handle.getProperty(PARTITION_TYPES_PROPERTY));
+ Type[] valueTypes = parseTypes((String)
handle.getProperty(VALUE_TYPES_PROPERTY));
+ String[] valueNames = decodeStrings((String)
handle.getProperty(VALUE_NAMES_PROPERTY));
+
+ return new TableFunctionProcessorProvider() {
+ @Override
+ public TableFunctionDataProcessor getDataProcessor() {
+ return new FFTDataProcessor(
+ sampleIntervalSpecified,
+ sampleInterval,
+ transformLength,
+ norm,
+ createColumns(partitionTypes, 1),
+ createNumericColumns(valueTypes, valueNames, partitionTypes.length
+ 1));
+ }
+ };
+ }
+
+ private static void validateOrderBy(TableArgument tableArgument, String
timeColumn) {
+ if (tableArgument.getOrderBy().size() != 1
+ || !tableArgument.getOrderBy().get(0).equalsIgnoreCase(timeColumn)) {
+ throw new
SemanticException(QueryMessages.FFT_ORDER_BY_MUST_CONTAIN_TIMECOL);
+ }
+ }
+
+ private static List<Integer> getPartitionIndexes(TableArgument tableArgument)
+ throws UDFException {
+ List<Integer> indexes = new ArrayList<>();
+ for (String partitionColumn : tableArgument.getPartitionBy()) {
+ indexes.add(findColumnIndex(tableArgument, partitionColumn,
SUPPORTED_PARTITION_TYPES));
+ }
+ return indexes;
+ }
+
+ public static void validateTransformLength(long transformLength, int
valueColumnCount) {
+ if (transformLength == UNSPECIFIED_N) {
+ return;
+ }
+ if (transformLength > MAX_TRANSFORM_LENGTH) {
+ throw new SemanticException(
+ String.format(QueryMessages.FFT_TRANSFORM_LENGTH_EXCEEDS_LIMIT,
MAX_TRANSFORM_LENGTH));
+ }
+ long spectrumValues;
+ try {
+ spectrumValues =
+ Math.multiplyExact(Math.multiplyExact(transformLength, 2L),
valueColumnCount);
+ } catch (ArithmeticException e) {
+ throw new SemanticException(QueryMessages.FFT_SPECTRUM_BUFFER_TOO_LARGE);
+ }
+ if (spectrumValues > MAX_SPECTRUM_VALUES) {
+ throw new SemanticException(QueryMessages.FFT_SPECTRUM_BUFFER_TOO_LARGE);
+ }
+ }
+
+ private static void validateNorm(String norm) {
+ if (!NORM_BACKWARD.equals(norm) && !NORM_FORWARD.equals(norm) &&
!NORM_ORTHO.equals(norm)) {
+ throw new SemanticException(QueryMessages.FFT_INVALID_NORM);
+ }
+ }
+
+ private static String joinTypes(List<Type> types) {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < types.size(); i++) {
+ if (i > 0) {
+ builder.append(',');
+ }
+ builder.append(types.get(i).name());
+ }
+ return builder.toString();
+ }
+
+ private static Type[] parseTypes(String value) {
+ if (value.isEmpty()) {
+ return new Type[0];
+ }
+ String[] values = value.split(",");
+ Type[] types = new Type[values.length];
+ for (int i = 0; i < values.length; i++) {
+ types[i] = Type.valueOf(values[i]);
+ }
+ return types;
+ }
+
+ private static String encodeStrings(List<String> values) {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < values.size(); i++) {
+ if (i > 0) {
+ builder.append(',');
+ }
+ builder.append(
+
Base64.getEncoder().encodeToString(values.get(i).getBytes(StandardCharsets.UTF_8)));
+ }
+ return builder.toString();
+ }
+
+ private static String[] decodeStrings(String value) {
+ if (value.isEmpty()) {
+ return new String[0];
+ }
+ String[] encodedValues = value.split(",");
+ String[] decodedValues = new String[encodedValues.length];
+ for (int i = 0; i < encodedValues.length; i++) {
+ decodedValues[i] =
+ new String(Base64.getDecoder().decode(encodedValues[i]),
StandardCharsets.UTF_8);
+ }
+ return decodedValues;
+ }
+
+ private static ValueColumn[] createColumns(Type[] types, int
firstInputIndex) {
+ ValueColumn[] columns = new ValueColumn[types.length];
+ for (int i = 0; i < types.length; i++) {
+ columns[i] = new ValueColumn(firstInputIndex + i,
ValueOperator.fromType(types[i]));
+ }
+ return columns;
+ }
+
+ private static NumericColumn[] createNumericColumns(
+ Type[] types, String[] names, int firstInputIndex) {
+ NumericColumn[] columns = new NumericColumn[types.length];
+ for (int i = 0; i < types.length; i++) {
+ columns[i] =
+ new NumericColumn(firstInputIndex + i, names[i],
NumericOperator.fromType(types[i]));
+ }
+ return columns;
+ }
+
+ private enum ValueOperator {
+ BOOLEAN(Type.BOOLEAN) {
+ @Override
+ Object read(Record record, int index) {
+ return record.getBoolean(index);
+ }
+
+ @Override
+ void write(ColumnBuilder builder, Object value) {
+ builder.writeBoolean((Boolean) value);
+ }
+ },
+ INT32(Type.INT32) {
+ @Override
+ Object read(Record record, int index) {
+ return record.getInt(index);
+ }
+
+ @Override
+ void write(ColumnBuilder builder, Object value) {
+ builder.writeInt((Integer) value);
+ }
+ },
+ INT64(Type.INT64) {
+ @Override
+ Object read(Record record, int index) {
+ return record.getLong(index);
+ }
+
+ @Override
+ void write(ColumnBuilder builder, Object value) {
+ builder.writeLong((Long) value);
+ }
+ },
+ FLOAT(Type.FLOAT) {
+ @Override
+ Object read(Record record, int index) {
+ return record.getFloat(index);
+ }
+
+ @Override
+ void write(ColumnBuilder builder, Object value) {
+ builder.writeFloat((Float) value);
+ }
+ },
+ DOUBLE(Type.DOUBLE) {
+ @Override
+ Object read(Record record, int index) {
+ return record.getDouble(index);
+ }
+
+ @Override
+ void write(ColumnBuilder builder, Object value) {
+ builder.writeDouble((Double) value);
+ }
+ },
+ TEXT(Type.TEXT) {
+ @Override
+ Object read(Record record, int index) {
+ return record.getBinary(index);
+ }
+
+ @Override
+ void write(ColumnBuilder builder, Object value) {
+ builder.writeBinary((Binary) value);
+ }
+ },
+ BLOB(Type.BLOB) {
+ @Override
+ Object read(Record record, int index) {
+ return record.getBinary(index);
+ }
+
+ @Override
+ void write(ColumnBuilder builder, Object value) {
+ builder.writeBinary((Binary) value);
+ }
+ },
+ TIMESTAMP(Type.TIMESTAMP) {
+ @Override
+ Object read(Record record, int index) {
+ return record.getLong(index);
+ }
+
+ @Override
+ void write(ColumnBuilder builder, Object value) {
+ builder.writeLong((Long) value);
+ }
+ },
+ DATE(Type.DATE) {
+ @Override
+ Object read(Record record, int index) {
+ return record.getLocalDate(index);
+ }
+
+ @Override
+ void write(ColumnBuilder builder, Object value) {
+ builder.writeObject(value);
+ }
+ },
+ STRING(Type.STRING) {
+ @Override
+ Object read(Record record, int index) {
+ return record.getBinary(index);
+ }
+
+ @Override
+ void write(ColumnBuilder builder, Object value) {
+ builder.writeBinary((Binary) value);
+ }
+ };
+
+ private final Type type;
+
+ ValueOperator(Type type) {
+ this.type = type;
+ }
+
+ abstract Object read(Record record, int index);
+
+ abstract void write(ColumnBuilder builder, Object value);
+
+ static ValueOperator fromType(Type type) {
+ for (ValueOperator valueOperator : values()) {
+ if (valueOperator.type == type) {
+ return valueOperator;
+ }
+ }
+ throw new IllegalArgumentException(
+ String.format(QueryMessages.FFT_UNSUPPORTED_PARTITION_TYPE, type));
+ }
+ }
+
+ private enum NumericOperator {
+ INT32(Type.INT32, false) {
+ @Override
+ Number read(Record record, int index) {
+ return record.getInt(index);
+ }
+ },
+ INT64(Type.INT64, false) {
+ @Override
+ Number read(Record record, int index) {
+ return record.getLong(index);
+ }
+ },
+ FLOAT(Type.FLOAT, true) {
+ @Override
+ Number read(Record record, int index) {
+ return record.getFloat(index);
+ }
+ },
+ DOUBLE(Type.DOUBLE, false) {
+ @Override
+ Number read(Record record, int index) {
+ return record.getDouble(index);
+ }
+ };
+
+ private final Type type;
+ private final boolean floatFft;
+
+ NumericOperator(Type type, boolean floatFft) {
+ this.type = type;
+ this.floatFft = floatFft;
+ }
+
+ abstract Number read(Record record, int index);
+
+ boolean usesFloatFft() {
+ return floatFft;
+ }
+
+ static NumericOperator fromType(Type type) {
+ for (NumericOperator numericOperator : values()) {
+ if (numericOperator.type == type) {
+ return numericOperator;
+ }
+ }
+ throw new IllegalArgumentException(
+ String.format(QueryMessages.FFT_UNSUPPORTED_VALUE_TYPE, type));
+ }
+ }
+
+ private static class ValueColumn {
+ private final int inputIndex;
+ private final ValueOperator valueOperator;
+
+ private ValueColumn(int inputIndex, ValueOperator valueOperator) {
+ this.inputIndex = inputIndex;
+ this.valueOperator = valueOperator;
+ }
+
+ private Object read(Record record) {
+ return valueOperator.read(record, inputIndex);
+ }
+
+ private void write(ColumnBuilder builder, Object value) {
+ valueOperator.write(builder, value);
+ }
+ }
+
+ private static class NumericColumn {
+ private final int inputIndex;
+ private final String name;
+ private final NumericOperator numericOperator;
+
+ private NumericColumn(int inputIndex, String name, NumericOperator
numericOperator) {
+ this.inputIndex = inputIndex;
+ this.name = name;
+ this.numericOperator = numericOperator;
+ }
+
+ private Number read(Record record) {
+ return numericOperator.read(record, inputIndex);
+ }
+
+ private boolean usesFloatFft() {
+ return numericOperator.usesFloatFft();
+ }
+ }
+
+ private static class Spectrum {
+ private final double[] doubleValues;
+ private final float[] floatValues;
+
+ private Spectrum(double[] doubleValues, float[] floatValues) {
+ this.doubleValues = doubleValues;
+ this.floatValues = floatValues;
+ }
+
+ private static Spectrum fromDouble(double[] values) {
+ return new Spectrum(values, null);
+ }
+
+ private static Spectrum fromFloat(float[] values) {
+ return new Spectrum(null, values);
+ }
+
+ private double real(int frequencyIndex, double scaleFactor) {
+ int index = 2 * frequencyIndex;
+ return (doubleValues == null ? floatValues[index] : doubleValues[index])
* scaleFactor;
+ }
+
+ private double imaginary(int frequencyIndex, double scaleFactor) {
+ int index = 2 * frequencyIndex + 1;
+ return (doubleValues == null ? floatValues[index] : doubleValues[index])
* scaleFactor;
+ }
+ }
+
+ private static class FFTDataProcessor implements TableFunctionDataProcessor {
+ private final boolean sampleIntervalSpecified;
+ private final long sampleInterval;
+ private final long specifiedTransformLength;
+ private final String norm;
+ private final ValueColumn[] partitionColumns;
+ private final NumericColumn[] valueColumns;
+ private final Object[] partitionValues;
+ private final boolean[] partitionValueIsNull;
+ private final List<Number[]> rows = new ArrayList<>();
+ private long inputRowCount;
+ private long firstTime;
+ private long previousTime;
+ private boolean initialized;
+
+ private FFTDataProcessor(
+ boolean sampleIntervalSpecified,
+ long sampleInterval,
+ long specifiedTransformLength,
+ String norm,
+ ValueColumn[] partitionColumns,
+ NumericColumn[] valueColumns) {
+ this.sampleIntervalSpecified = sampleIntervalSpecified;
+ this.sampleInterval = sampleInterval;
+ this.specifiedTransformLength = specifiedTransformLength;
+ this.norm = norm;
+ this.partitionColumns = partitionColumns;
+ this.valueColumns = valueColumns;
+ this.partitionValues = new Object[partitionColumns.length];
+ this.partitionValueIsNull = new boolean[partitionColumns.length];
+ }
+
+ @Override
+ public void process(
+ Record input,
+ List<ColumnBuilder> properColumnBuilders,
+ ColumnBuilder passThroughIndexBuilder) {
+ long currentTime = input.getLong(0);
+ if (!initialized) {
+ capturePartitionValues(input);
+ firstTime = currentTime;
+ initialized = true;
+ } else if (currentTime <= previousTime) {
+ throw new
SemanticException(QueryMessages.FFT_TIME_MUST_BE_STRICTLY_ASCENDING);
+ }
+ previousTime = currentTime;
+
+ if (specifiedTransformLength == UNSPECIFIED_N) {
+ validateTransformLength(inputRowCount + 1, valueColumns.length);
+ }
+
+ boolean shouldCacheRow =
+ specifiedTransformLength == UNSPECIFIED_N || rows.size() <
specifiedTransformLength;
+ Number[] row = shouldCacheRow ? new Number[valueColumns.length] : null;
+ for (int i = 0; i < valueColumns.length; i++) {
+ NumericColumn valueColumn = valueColumns[i];
+ if (input.isNull(valueColumn.inputIndex)) {
+ throw new SemanticException(
+ String.format(QueryMessages.FFT_NULL_VALUE_NOT_SUPPORTED,
valueColumn.name));
+ }
+ if (shouldCacheRow) {
+ row[i] = valueColumn.read(input);
+ }
+ }
+ inputRowCount++;
+ if (shouldCacheRow) {
+ rows.add(row);
+ }
+ }
+
+ @Override
+ public void finish(
+ List<ColumnBuilder> properColumnBuilders, ColumnBuilder
passThroughIndexBuilder) {
+ if (inputRowCount == 0) {
+ return;
+ }
+
+ int transformLength = getTransformLength();
+ double sampleIntervalSeconds = getSampleIntervalSeconds();
+ double scaleFactor = getScaleFactor(transformLength);
+ Spectrum[] spectra = new Spectrum[valueColumns.length];
+
+ int copiedRows = Math.min(rows.size(), transformLength);
+ DoubleFFT_1D doubleFft = null;
+ FloatFFT_1D floatFft = null;
+ for (int columnIndex = 0; columnIndex < valueColumns.length;
columnIndex++) {
+ if (valueColumns[columnIndex].usesFloatFft()) {
+ float[] spectrum = new float[2 * transformLength];
+ for (int rowIndex = 0; rowIndex < copiedRows; rowIndex++) {
+ spectrum[2 * rowIndex] =
rows.get(rowIndex)[columnIndex].floatValue();
+ }
+ if (floatFft == null) {
+ floatFft = new FloatFFT_1D(transformLength);
+ }
+ floatFft.complexForward(spectrum);
+ spectra[columnIndex] = Spectrum.fromFloat(spectrum);
+ } else {
+ double[] spectrum = new double[2 * transformLength];
+ for (int rowIndex = 0; rowIndex < copiedRows; rowIndex++) {
+ spectrum[2 * rowIndex] =
rows.get(rowIndex)[columnIndex].doubleValue();
+ }
+ if (doubleFft == null) {
+ doubleFft = new DoubleFFT_1D(transformLength);
+ }
+ doubleFft.complexForward(spectrum);
+ spectra[columnIndex] = Spectrum.fromDouble(spectrum);
+ }
+ }
+
+ for (int frequencyIndex = 0; frequencyIndex < transformLength;
frequencyIndex++) {
+ int outputColumnIndex = 0;
+ for (int partitionIndex = 0; partitionIndex < partitionColumns.length;
partitionIndex++) {
+ if (partitionValueIsNull[partitionIndex]) {
+ properColumnBuilders.get(outputColumnIndex++).appendNull();
+ } else {
+ partitionColumns[partitionIndex].write(
+ properColumnBuilders.get(outputColumnIndex++),
partitionValues[partitionIndex]);
+ }
+ }
+
properColumnBuilders.get(outputColumnIndex++).writeLong(frequencyIndex);
+ properColumnBuilders
+ .get(outputColumnIndex++)
+ .writeDouble(
+ calculateFrequency(frequencyIndex, transformLength,
sampleIntervalSeconds));
+ for (int columnIndex = 0; columnIndex < valueColumns.length;
columnIndex++) {
+ properColumnBuilders
+ .get(outputColumnIndex++)
+ .writeDouble(spectra[columnIndex].real(frequencyIndex,
scaleFactor));
+ properColumnBuilders
+ .get(outputColumnIndex++)
+ .writeDouble(spectra[columnIndex].imaginary(frequencyIndex,
scaleFactor));
+ }
+ }
+ }
+
+ private void capturePartitionValues(Record input) {
+ for (int i = 0; i < partitionColumns.length; i++) {
+ if (input.isNull(partitionColumns[i].inputIndex)) {
+ partitionValueIsNull[i] = true;
+ } else {
+ partitionValues[i] = partitionColumns[i].read(input);
+ }
+ }
+ }
+
+ private int getTransformLength() {
+ long transformLength =
+ specifiedTransformLength == UNSPECIFIED_N ? inputRowCount :
specifiedTransformLength;
+ validateTransformLength(transformLength, valueColumns.length);
+ return (int) transformLength;
+ }
+
+ private double getSampleIntervalSeconds() {
+ double interval;
+ if (sampleIntervalSpecified) {
+ interval = sampleInterval;
+ } else {
+ if (inputRowCount < 2) {
+ throw new
SemanticException(QueryMessages.FFT_NEEDS_TWO_ROWS_FOR_INTERVAL);
+ }
+ // Convert before subtracting so nanosecond timestamps spanning more
than Long.MAX_VALUE
+ // do not overflow as longs.
+ interval = ((double) previousTime - (double) firstTime) /
(inputRowCount - 1);
+ }
+ double intervalSeconds =
+ interval * TimestampPrecisionUtils.currPrecision.toNanos(1L) /
1_000_000_000.0;
+ if (intervalSeconds <= 0) {
+ throw new
SemanticException(QueryMessages.FFT_SAMPLE_INTERVAL_MUST_BE_POSITIVE);
+ }
+ return intervalSeconds;
+ }
+
+ private double getScaleFactor(int transformLength) {
+ if (NORM_FORWARD.equals(norm)) {
+ return 1.0 / transformLength;
+ }
+ if (NORM_ORTHO.equals(norm)) {
+ return 1.0 / Math.sqrt(transformLength);
+ }
+ return 1.0;
+ }
+
+ private double calculateFrequency(
+ int frequencyIndex, int transformLength, double sampleIntervalSeconds)
{
+ int positiveFrequencyCount = (transformLength + 1) / 2;
+ int signedIndex =
+ frequencyIndex < positiveFrequencyCount
+ ? frequencyIndex
+ : frequencyIndex - transformLength;
+ return signedIndex / (transformLength * sampleIntervalSeconds);
+ }
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/DoubleFFT_1D.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/DoubleFFT_1D.java
new file mode 100644
index 00000000000..e0d6b89f1ba
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/DoubleFFT_1D.java
@@ -0,0 +1,159 @@
+/*
+ * 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.builtin.relational.tvf.fft;
+
+import org.apache.iotdb.commons.i18n.QueryMessages;
+
+/** Computes an in-place 1D forward DFT for interleaved complex double data. */
+public final class DoubleFFT_1D {
+
+ private final int length;
+
+ public DoubleFFT_1D(long length) {
+ if (length < 1 || length > Integer.MAX_VALUE) {
+ throw new IllegalArgumentException(
+
QueryMessages.EXCEPTION_FFT_LENGTH_MUST_BE_A_POSITIVE_INT_SIZED_VALUE_A000D3BB);
+ }
+ this.length = (int) length;
+ }
+
+ public void complexForward(double[] values) {
+ if (values.length < 2 * length) {
+ throw new IllegalArgumentException(
+
QueryMessages.EXCEPTION_INPUT_ARRAY_LENGTH_MUST_BE_AT_LEAST_2_FFT_LENGTH_31DF6A25);
+ }
+ if (length == 1) {
+ return;
+ }
+ if (isPowerOfTwo(length)) {
+ transform(values, length, false);
+ } else {
+ bluesteinForward(values);
+ }
+ }
+
+ private void bluesteinForward(double[] values) {
+ int convolutionLength = nextPowerOfTwo(2 * length - 1);
+ double[] a = new double[2 * convolutionLength];
+ double[] b = new double[2 * convolutionLength];
+
+ for (int i = 0; i < length; i++) {
+ double angle = Math.PI * i * (double) i / length;
+ double cos = Math.cos(angle);
+ double sin = Math.sin(angle);
+ double real = values[2 * i];
+ double imaginary = values[2 * i + 1];
+
+ a[2 * i] = real * cos + imaginary * sin;
+ a[2 * i + 1] = imaginary * cos - real * sin;
+ b[2 * i] = cos;
+ b[2 * i + 1] = sin;
+ if (i > 0) {
+ b[2 * (convolutionLength - i)] = cos;
+ b[2 * (convolutionLength - i) + 1] = sin;
+ }
+ }
+
+ transform(a, convolutionLength, false);
+ transform(b, convolutionLength, false);
+ for (int i = 0; i < convolutionLength; i++) {
+ int offset = 2 * i;
+ double real = a[offset] * b[offset] - a[offset + 1] * b[offset + 1];
+ double imaginary = a[offset] * b[offset + 1] + a[offset + 1] * b[offset];
+ a[offset] = real;
+ a[offset + 1] = imaginary;
+ }
+ transform(a, convolutionLength, true);
+
+ for (int i = 0; i < length; i++) {
+ double angle = Math.PI * i * (double) i / length;
+ double cos = Math.cos(angle);
+ double sin = Math.sin(angle);
+ double real = a[2 * i];
+ double imaginary = a[2 * i + 1];
+ values[2 * i] = real * cos + imaginary * sin;
+ values[2 * i + 1] = imaginary * cos - real * sin;
+ }
+ }
+
+ private static void transform(double[] values, int size, boolean inverse) {
+ for (int i = 1, j = 0; i < size; i++) {
+ int bit = size >>> 1;
+ while ((j & bit) != 0) {
+ j ^= bit;
+ bit >>>= 1;
+ }
+ j ^= bit;
+ if (i < j) {
+ swap(values, 2 * i, 2 * j);
+ swap(values, 2 * i + 1, 2 * j + 1);
+ }
+ }
+
+ for (int step = 2; step <= size; step <<= 1) {
+ double angle = (inverse ? 2.0 : -2.0) * Math.PI / step;
+ double stepReal = Math.cos(angle);
+ double stepImaginary = Math.sin(angle);
+ int halfStep = step >>> 1;
+ for (int block = 0; block < size; block += step) {
+ double factorReal = 1.0;
+ double factorImaginary = 0.0;
+ for (int j = 0; j < halfStep; j++) {
+ int even = 2 * (block + j);
+ int odd = 2 * (block + j + halfStep);
+ double oddReal = values[odd] * factorReal - values[odd + 1] *
factorImaginary;
+ double oddImaginary = values[odd] * factorImaginary + values[odd +
1] * factorReal;
+ double evenReal = values[even];
+ double evenImaginary = values[even + 1];
+
+ values[even] = evenReal + oddReal;
+ values[even + 1] = evenImaginary + oddImaginary;
+ values[odd] = evenReal - oddReal;
+ values[odd + 1] = evenImaginary - oddImaginary;
+
+ double nextFactorReal = factorReal * stepReal - factorImaginary *
stepImaginary;
+ factorImaginary = factorReal * stepImaginary + factorImaginary *
stepReal;
+ factorReal = nextFactorReal;
+ }
+ }
+ }
+
+ if (inverse) {
+ for (int i = 0; i < 2 * size; i++) {
+ values[i] /= size;
+ }
+ }
+ }
+
+ private static boolean isPowerOfTwo(int value) {
+ return (value & (value - 1)) == 0;
+ }
+
+ private static int nextPowerOfTwo(int value) {
+ int highestOneBit = Integer.highestOneBit(value);
+ return value == highestOneBit ? value : highestOneBit << 1;
+ }
+
+ private static void swap(double[] values, int left, int right) {
+ double tmp = values[left];
+ values[left] = values[right];
+ values[right] = tmp;
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FloatFFT_1D.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FloatFFT_1D.java
new file mode 100644
index 00000000000..7828197f0b9
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FloatFFT_1D.java
@@ -0,0 +1,160 @@
+/*
+ * 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.builtin.relational.tvf.fft;
+
+import org.apache.iotdb.commons.i18n.QueryMessages;
+
+/** Computes an in-place 1D forward DFT for interleaved complex float data. */
+public final class FloatFFT_1D {
+
+ private final int length;
+
+ public FloatFFT_1D(long length) {
+ if (length < 1 || length > Integer.MAX_VALUE) {
+ throw new IllegalArgumentException(
+
QueryMessages.EXCEPTION_FFT_LENGTH_MUST_BE_A_POSITIVE_INT_SIZED_VALUE_A000D3BB);
+ }
+ this.length = (int) length;
+ }
+
+ public void complexForward(float[] values) {
+ if (values.length < 2 * length) {
+ throw new IllegalArgumentException(
+
QueryMessages.EXCEPTION_INPUT_ARRAY_LENGTH_MUST_BE_AT_LEAST_2_FFT_LENGTH_31DF6A25);
+ }
+ if (length == 1) {
+ return;
+ }
+ if (isPowerOfTwo(length)) {
+ transform(values, length, false);
+ } else {
+ bluesteinForward(values);
+ }
+ }
+
+ private void bluesteinForward(float[] values) {
+ int convolutionLength = nextPowerOfTwo(2 * length - 1);
+ float[] a = new float[2 * convolutionLength];
+ float[] b = new float[2 * convolutionLength];
+
+ for (int i = 0; i < length; i++) {
+ double angle = Math.PI * i * (double) i / length;
+ float cos = (float) Math.cos(angle);
+ float sin = (float) Math.sin(angle);
+ float real = values[2 * i];
+ float imaginary = values[2 * i + 1];
+
+ a[2 * i] = real * cos + imaginary * sin;
+ a[2 * i + 1] = imaginary * cos - real * sin;
+ b[2 * i] = cos;
+ b[2 * i + 1] = sin;
+ if (i > 0) {
+ b[2 * (convolutionLength - i)] = cos;
+ b[2 * (convolutionLength - i) + 1] = sin;
+ }
+ }
+
+ transform(a, convolutionLength, false);
+ transform(b, convolutionLength, false);
+ for (int i = 0; i < convolutionLength; i++) {
+ int offset = 2 * i;
+ float real = a[offset] * b[offset] - a[offset + 1] * b[offset + 1];
+ float imaginary = a[offset] * b[offset + 1] + a[offset + 1] * b[offset];
+ a[offset] = real;
+ a[offset + 1] = imaginary;
+ }
+ transform(a, convolutionLength, true);
+
+ for (int i = 0; i < length; i++) {
+ double angle = Math.PI * i * (double) i / length;
+ float cos = (float) Math.cos(angle);
+ float sin = (float) Math.sin(angle);
+ float real = a[2 * i];
+ float imaginary = a[2 * i + 1];
+ values[2 * i] = real * cos + imaginary * sin;
+ values[2 * i + 1] = imaginary * cos - real * sin;
+ }
+ }
+
+ private static void transform(float[] values, int size, boolean inverse) {
+ for (int i = 1, j = 0; i < size; i++) {
+ int bit = size >>> 1;
+ while ((j & bit) != 0) {
+ j ^= bit;
+ bit >>>= 1;
+ }
+ j ^= bit;
+ if (i < j) {
+ swap(values, 2 * i, 2 * j);
+ swap(values, 2 * i + 1, 2 * j + 1);
+ }
+ }
+
+ for (int step = 2; step <= size; step <<= 1) {
+ double angle = (inverse ? 2.0 : -2.0) * Math.PI / step;
+ double stepReal = Math.cos(angle);
+ double stepImaginary = Math.sin(angle);
+ int halfStep = step >>> 1;
+ for (int block = 0; block < size; block += step) {
+ double factorReal = 1.0;
+ double factorImaginary = 0.0;
+ for (int j = 0; j < halfStep; j++) {
+ int even = 2 * (block + j);
+ int odd = 2 * (block + j + halfStep);
+ float oddReal = (float) (values[odd] * factorReal - values[odd + 1]
* factorImaginary);
+ float oddImaginary =
+ (float) (values[odd] * factorImaginary + values[odd + 1] *
factorReal);
+ float evenReal = values[even];
+ float evenImaginary = values[even + 1];
+
+ values[even] = evenReal + oddReal;
+ values[even + 1] = evenImaginary + oddImaginary;
+ values[odd] = evenReal - oddReal;
+ values[odd + 1] = evenImaginary - oddImaginary;
+
+ double nextFactorReal = factorReal * stepReal - factorImaginary *
stepImaginary;
+ factorImaginary = factorReal * stepImaginary + factorImaginary *
stepReal;
+ factorReal = nextFactorReal;
+ }
+ }
+ }
+
+ if (inverse) {
+ for (int i = 0; i < 2 * size; i++) {
+ values[i] /= size;
+ }
+ }
+ }
+
+ private static boolean isPowerOfTwo(int value) {
+ return (value & (value - 1)) == 0;
+ }
+
+ private static int nextPowerOfTwo(int value) {
+ int highestOneBit = Integer.highestOneBit(value);
+ return value == highestOneBit ? value : highestOneBit << 1;
+ }
+
+ private static void swap(float[] values, int left, int right) {
+ float tmp = values[left];
+ values[left] = values[right];
+ values[right] = tmp;
+ }
+}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunctionTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunctionTest.java
new file mode 100644
index 00000000000..6dcef223c1d
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunctionTest.java
@@ -0,0 +1,518 @@
+/*
+ * 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.builtin.relational.tvf;
+
+import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils;
+import org.apache.iotdb.udf.api.exception.UDFException;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis;
+import org.apache.iotdb.udf.api.relational.table.TableFunctionHandle;
+import org.apache.iotdb.udf.api.relational.table.argument.Argument;
+import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
+import org.apache.iotdb.udf.api.relational.table.argument.TableArgument;
+import
org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor;
+import org.apache.iotdb.udf.api.type.Type;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.read.common.block.column.DoubleColumnBuilder;
+import org.apache.tsfile.read.common.block.column.LongColumnBuilder;
+import org.apache.tsfile.utils.Binary;
+import org.junit.Test;
+
+import java.io.File;
+import java.time.LocalDate;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+public class FFTTableFunctionTest {
+
+ private static final double DELTA = 1e-9;
+
+ private final FFTTableFunction function = new FFTTableFunction();
+
+ @Test
+ public void testWritesFullSpectrumAndZeroPadsToSpecifiedN() throws
UDFException {
+ TableFunctionDataProcessor processor = createProcessor(true, 4L);
+ processor.process(record(0L, 1.0), Collections.emptyList(), null);
+
+ List<ColumnBuilder> builders = createOutputBuilders(4);
+ processor.finish(builders, null);
+
+ double intervalSeconds = TimestampPrecisionUtils.currPrecision.toNanos(1L)
/ 1_000_000_000.0;
+ assertLongColumn(builders.get(0).build(), 0L, 1L, 2L, 3L);
+ assertDoubleColumn(
+ builders.get(1).build(),
+ 0.0,
+ 1.0 / (4.0 * intervalSeconds),
+ -2.0 / (4.0 * intervalSeconds),
+ -1.0 / (4.0 * intervalSeconds));
+ assertDoubleColumn(builders.get(2).build(), 1.0, 1.0, 1.0, 1.0);
+ assertDoubleColumn(builders.get(3).build(), 0.0, 0.0, 0.0, 0.0);
+ }
+
+ @Test
+ public void testTruncatesInputRowsToSpecifiedN() throws UDFException {
+ TableFunctionDataProcessor processor = createProcessor(true, 2L);
+ processor.process(record(0L, 1.0), Collections.emptyList(), null);
+ processor.process(record(1L, 2.0), Collections.emptyList(), null);
+ processor.process(record(2L, 100.0), Collections.emptyList(), null);
+ processor.process(record(3L, 200.0), Collections.emptyList(), null);
+
+ List<ColumnBuilder> builders = createOutputBuilders(2);
+ processor.finish(builders, null);
+
+ assertLongColumn(builders.get(0).build(), 0L, 1L);
+ assertDoubleColumn(builders.get(2).build(), 3.0, -1.0);
+ assertDoubleColumn(builders.get(3).build(), 0.0, 0.0);
+ }
+
+ @Test
+ public void testSupportsNonPowerOfTwoTransformLength() throws UDFException {
+ TableFunctionDataProcessor processor = createProcessor(true, 3L);
+ processor.process(record(0L, 1.0), Collections.emptyList(), null);
+ processor.process(record(1L, 2.0), Collections.emptyList(), null);
+ processor.process(record(2L, 3.0), Collections.emptyList(), null);
+
+ List<ColumnBuilder> builders = createOutputBuilders(3);
+ processor.finish(builders, null);
+
+ double imaginaryComponent = Math.sqrt(3.0) / 2.0;
+ assertLongColumn(builders.get(0).build(), 0L, 1L, 2L);
+ assertDoubleColumn(builders.get(2).build(), 6.0, -1.5, -1.5);
+ assertDoubleColumn(builders.get(3).build(), 0.0, imaginaryComponent,
-imaginaryComponent);
+ }
+
+ @Test
+ public void testSupportsFloatInputWithFloatFft() throws UDFException {
+ TableFunctionDataProcessor processor = createProcessor(Type.FLOAT, true,
3L);
+ processor.process(record(0L, 1.0f), Collections.emptyList(), null);
+ processor.process(record(1L, 2.0f), Collections.emptyList(), null);
+ processor.process(record(2L, 3.0f), Collections.emptyList(), null);
+
+ List<ColumnBuilder> builders = createOutputBuilders(3);
+ processor.finish(builders, null);
+
+ double imaginaryComponent = Math.sqrt(3.0) / 2.0;
+ assertLongColumn(builders.get(0).build(), 0L, 1L, 2L);
+ assertDoubleColumnWithDelta(builders.get(2).build(), 1e-6, 6.0, -1.5,
-1.5);
+ assertDoubleColumnWithDelta(
+ builders.get(3).build(), 1e-6, 0.0, imaginaryComponent,
-imaginaryComponent);
+ }
+
+ @Test
+ public void testRejectsInvalidRowsEvenWhenBeyondTruncatedN() throws
UDFException {
+ TableFunctionDataProcessor processor = createProcessor(true, 2L);
+ processor.process(record(0L, 1.0), Collections.emptyList(), null);
+ processor.process(record(1L, 2.0), Collections.emptyList(), null);
+
+ assertSemanticException(
+ () -> processor.process(nullValueRecord(2L), Collections.emptyList(),
null),
+ "FFT does not support null values in column [value].");
+ }
+
+ @Test
+ public void testInfersSampleIntervalFromFullInputRangeBeyondSpecifiedN()
throws UDFException {
+ TableFunctionDataProcessor processor = createProcessor(false, 2L);
+ processor.process(record(0L, 1.0), Collections.emptyList(), null);
+ processor.process(record(1L, 2.0), Collections.emptyList(), null);
+ processor.process(record(3L, 3.0), Collections.emptyList(), null);
+
+ List<ColumnBuilder> builders = createOutputBuilders(2);
+ processor.finish(builders, null);
+
+ double intervalSeconds =
+ 1.5 * TimestampPrecisionUtils.currPrecision.toNanos(1L) /
1_000_000_000.0;
+ assertLongColumn(builders.get(0).build(), 0L, 1L);
+ assertDoubleColumn(builders.get(1).build(), 0.0, -1.0 / (2.0 *
intervalSeconds));
+ assertDoubleColumn(builders.get(2).build(), 3.0, -1.0);
+ assertDoubleColumn(builders.get(3).build(), 0.0, 0.0);
+ }
+
+ @Test
+ public void testRejectsDuplicateTime() throws UDFException {
+ TableFunctionDataProcessor processor = createProcessor(false);
+ processor.process(record(1L, 1.0), Collections.emptyList(), null);
+
+ assertSemanticException(
+ () -> processor.process(record(1L, 2.0), Collections.emptyList(),
null),
+ "The time column of FFT input must be strictly ascending within each
partition.");
+ }
+
+ @Test
+ public void testRejectsOutOfOrderTime() throws UDFException {
+ TableFunctionDataProcessor processor = createProcessor(false);
+ processor.process(record(2L, 1.0), Collections.emptyList(), null);
+
+ assertSemanticException(
+ () -> processor.process(record(1L, 2.0), Collections.emptyList(),
null),
+ "The time column of FFT input must be strictly ascending within each
partition.");
+ }
+
+ @Test
+ public void testRejectsSingleRowWithoutSampleInterval() throws UDFException {
+ TableFunctionDataProcessor processor = createProcessor(false);
+ processor.process(record(1L, 1.0), Collections.emptyList(), null);
+
+ assertSemanticException(
+ () -> processor.finish(Collections.emptyList(), null),
+ "FFT requires at least two rows to infer SAMPLE_INTERVAL.");
+ }
+
+ @Test
+ public void testInfersSampleIntervalFromPartitionTimeRange() throws
UDFException {
+ TableFunctionDataProcessor processor = createProcessor(false);
+ processor.process(record(0L, 1.0), Collections.emptyList(), null);
+ processor.process(record(1L, 2.0), Collections.emptyList(), null);
+ processor.process(record(3L, 3.0), Collections.emptyList(), null);
+
+ List<ColumnBuilder> builders = createOutputBuilders(3);
+ processor.finish(builders, null);
+
+ double intervalSeconds =
+ 1.5 * TimestampPrecisionUtils.currPrecision.toNanos(1L) /
1_000_000_000.0;
+ assertLongColumn(builders.get(0).build(), 0L, 1L, 2L);
+ assertDoubleColumn(
+ builders.get(1).build(),
+ 0.0,
+ 1.0 / (3.0 * intervalSeconds),
+ -1.0 / (3.0 * intervalSeconds));
+ }
+
+ @Test
+ public void testInfersIntervalWithoutNanosecondTimestampOverflow() throws
UDFException {
+ TableFunctionDataProcessor processor = createProcessor(false, 2L);
+ processor.process(record(-6_000_000_000_000_000_000L, 1.0),
Collections.emptyList(), null);
+ processor.process(record(6_000_000_000_000_000_000L, 2.0),
Collections.emptyList(), null);
+
+ List<ColumnBuilder> builders = createOutputBuilders(2);
+ processor.finish(builders, null);
+
+ double intervalSeconds =
+ 12.0e18 * TimestampPrecisionUtils.currPrecision.toNanos(1L) /
1_000_000_000.0;
+ assertDoubleColumn(builders.get(1).build(), 0.0, -1.0 / (2.0 *
intervalSeconds));
+ }
+
+ @Test
+ public void testUsesExplicitSampleIntervalWithoutGapValidation() throws
UDFException {
+ TableFunctionDataProcessor processor = createProcessor(true);
+ processor.process(record(0L, 1.0), Collections.emptyList(), null);
+ processor.process(record(2L, 2.0), Collections.emptyList(), null);
+
+ List<ColumnBuilder> builders = createOutputBuilders(2);
+ processor.finish(builders, null);
+
+ double intervalSeconds = TimestampPrecisionUtils.currPrecision.toNanos(1L)
/ 1_000_000_000.0;
+ assertLongColumn(builders.get(0).build(), 0L, 1L);
+ assertDoubleColumn(builders.get(1).build(), 0.0, -1.0 / (2.0 *
intervalSeconds));
+ }
+
+ @Test
+ public void testRejectsDefaultTransformLengthAboveLimit() throws
UDFException {
+ TableFunctionDataProcessor processor = createProcessor(true);
+ for (long time = 0; time < 65_536L; time++) {
+ processor.process(record(time, 1.0), Collections.emptyList(), null);
+ }
+
+ assertSemanticException(
+ () -> processor.process(record(65_536L, 1.0), Collections.emptyList(),
null),
+ "FFT transform length N must not exceed 65536.");
+ }
+
+ @Test
+ public void testRejectsDefaultSpectrumBufferAboveLimit() {
+ assertSemanticException(
+ () -> FFTTableFunction.validateTransformLength(65_536L, 129),
+ "FFT spectrum buffer is too large. Reduce N or the number of numeric
columns.");
+ }
+
+ @Test
+ public void testAnalyzeUsesSpecifiedTimeColumn() throws UDFException {
+ Map<String, Argument> arguments = createArguments("event_time",
"event_time");
+
+ TableFunctionAnalysis analysis = function.analyze(arguments);
+
+ assertEquals(
+ Arrays.asList(0, 1),
+
analysis.getRequiredColumns().get(FFTTableFunction.DATA_PARAMETER_NAME));
+ assertEquals(
+ "value_real",
analysis.getProperColumnSchema().get().getFields().get(2).getName().get());
+ assertEquals(
+ "value_imag",
analysis.getProperColumnSchema().get().getFields().get(3).getName().get());
+ }
+
+ @Test
+ public void testAnalyzePreservesValueColumnNamesContainingComma() throws
UDFException {
+ Map<String, Argument> arguments = createArguments("time", "time",
"value,with,comma");
+
+ TableFunctionAnalysis analysis = function.analyze(arguments);
+
+ assertEquals(
+ "value,with,comma_real",
+
analysis.getProperColumnSchema().get().getFields().get(2).getName().get());
+ assertEquals(
+ "value,with,comma_imag",
+
analysis.getProperColumnSchema().get().getFields().get(3).getName().get());
+
+ TableFunctionHandle handle = function.createTableFunctionHandle();
+ handle.deserialize(analysis.getTableFunctionHandle().serialize());
+ TableFunctionDataProcessor processor =
function.getProcessorProvider(handle).getDataProcessor();
+ processor.process(record(0L, 1.0), Collections.emptyList(), null);
+ assertSemanticException(
+ () -> processor.process(nullValueRecord(1L), Collections.emptyList(),
null),
+ "FFT does not support null values in column [value,with,comma].");
+ }
+
+ @Test
+ public void testAnalyzeRejectsOrderByDifferentFromSpecifiedTimeColumn() {
+ assertSemanticException(
+ () -> {
+ try {
+ function.analyze(createArguments("event_time", "time"));
+ } catch (UDFException e) {
+ throw new AssertionError(e);
+ }
+ },
+ "The ORDER BY clause of the DATA argument must contain exactly the
time column specified by the TIMECOL argument.");
+ }
+
+ private TableFunctionDataProcessor createProcessor(boolean
sampleIntervalSpecified)
+ throws UDFException {
+ return createProcessor(sampleIntervalSpecified, -1L);
+ }
+
+ private TableFunctionDataProcessor createProcessor(
+ boolean sampleIntervalSpecified, long transformLength) throws
UDFException {
+ return createProcessor(Type.DOUBLE, sampleIntervalSpecified,
transformLength);
+ }
+
+ private TableFunctionDataProcessor createProcessor(
+ Type valueType, boolean sampleIntervalSpecified, long transformLength)
throws UDFException {
+ Map<String, Argument> arguments = new HashMap<>();
+ arguments.put(
+ FFTTableFunction.DATA_PARAMETER_NAME,
+ new TableArgument(
+ Arrays.asList(Optional.of("time"), Optional.of("value")),
+ Arrays.asList(Type.TIMESTAMP, valueType),
+ Collections.emptyList(),
+ Collections.singletonList("time"),
+ false));
+ arguments.put(FFTTableFunction.TIMECOL_PARAMETER_NAME, new
ScalarArgument(Type.STRING, "time"));
+ arguments.put(
+ FFTTableFunction.SAMPLE_INTERVAL_PARAMETER_NAME,
+ new ScalarArgument(Type.INT64, sampleIntervalSpecified ? 1L :
Long.MIN_VALUE));
+ arguments.put(
+ FFTTableFunction.SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME,
+ new ScalarArgument(Type.BOOLEAN, sampleIntervalSpecified));
+ arguments.put(
+ FFTTableFunction.N_PARAMETER_NAME, new ScalarArgument(Type.INT64,
transformLength));
+ arguments.put(
+ FFTTableFunction.NORM_PARAMETER_NAME, new ScalarArgument(Type.STRING,
"backward"));
+
+ return function
+
.getProcessorProvider(function.analyze(arguments).getTableFunctionHandle())
+ .getDataProcessor();
+ }
+
+ private Map<String, Argument> createArguments(String timeColumn, String
orderByColumn) {
+ return createArguments(timeColumn, orderByColumn, "value");
+ }
+
+ private Map<String, Argument> createArguments(
+ String timeColumn, String orderByColumn, String valueColumnName) {
+ Map<String, Argument> arguments = new HashMap<>();
+ arguments.put(
+ FFTTableFunction.DATA_PARAMETER_NAME,
+ new TableArgument(
+ Arrays.asList(Optional.of(timeColumn),
Optional.of(valueColumnName)),
+ Arrays.asList(Type.TIMESTAMP, Type.DOUBLE),
+ Collections.emptyList(),
+ Collections.singletonList(orderByColumn),
+ false));
+ arguments.put(
+ FFTTableFunction.TIMECOL_PARAMETER_NAME, new
ScalarArgument(Type.STRING, timeColumn));
+ arguments.put(
+ FFTTableFunction.SAMPLE_INTERVAL_PARAMETER_NAME, new
ScalarArgument(Type.INT64, 1L));
+ arguments.put(
+ FFTTableFunction.SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME,
+ new ScalarArgument(Type.BOOLEAN, true));
+ arguments.put(FFTTableFunction.N_PARAMETER_NAME, new
ScalarArgument(Type.INT64, -1L));
+ arguments.put(
+ FFTTableFunction.NORM_PARAMETER_NAME, new ScalarArgument(Type.STRING,
"backward"));
+ return arguments;
+ }
+
+ private Record record(long time, double value) {
+ return new SimpleRecord(time, value);
+ }
+
+ private Record record(long time, float value) {
+ return new SimpleRecord(time, value);
+ }
+
+ private Record nullValueRecord(long time) {
+ return new SimpleRecord(time, null);
+ }
+
+ private List<ColumnBuilder> createOutputBuilders(int expectedPositionCount) {
+ return Arrays.asList(
+ new LongColumnBuilder(null, expectedPositionCount),
+ new DoubleColumnBuilder(null, expectedPositionCount),
+ new DoubleColumnBuilder(null, expectedPositionCount),
+ new DoubleColumnBuilder(null, expectedPositionCount));
+ }
+
+ private void assertLongColumn(Column column, long... expected) {
+ assertEquals(expected.length, column.getPositionCount());
+ for (int i = 0; i < expected.length; i++) {
+ assertEquals(expected[i], column.getLong(i));
+ }
+ }
+
+ private void assertDoubleColumn(Column column, double... expected) {
+ assertDoubleColumnWithDelta(column, DELTA, expected);
+ }
+
+ private void assertDoubleColumnWithDelta(Column column, double delta,
double... expected) {
+ assertEquals(expected.length, column.getPositionCount());
+ for (int i = 0; i < expected.length; i++) {
+ assertEquals(expected[i], column.getDouble(i), delta);
+ }
+ }
+
+ private void assertSemanticException(Runnable runnable, String message) {
+ try {
+ runnable.run();
+ fail();
+ } catch (SemanticException e) {
+ assertEquals(message, e.getMessage());
+ }
+ }
+
+ private static class SimpleRecord implements Record {
+ private final long time;
+ private final Number value;
+
+ private SimpleRecord(long time, Number value) {
+ this.time = time;
+ this.value = value;
+ }
+
+ @Override
+ public int getInt(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getLong(int columnIndex) {
+ if (columnIndex == 0) {
+ return time;
+ }
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public float getFloat(int columnIndex) {
+ if (columnIndex == 1 && value instanceof Float) {
+ return value.floatValue();
+ }
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public double getDouble(int columnIndex) {
+ if (columnIndex == 1 && value instanceof Double) {
+ return value.doubleValue();
+ }
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean getBoolean(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Binary getBinary(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String getString(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public LocalDate getLocalDate(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Object getObject(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional<File> getObjectFile(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long objectLength(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Binary readObject(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Binary readObject(int columnIndex, long offset, int length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Type getDataType(int columnIndex) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isNull(int columnIndex) {
+ if (columnIndex == 1) {
+ return value == null;
+ }
+ return false;
+ }
+
+ @Override
+ public int size() {
+ return 2;
+ }
+ }
+}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FFT1DTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FFT1DTest.java
new file mode 100644
index 00000000000..86f9d6c11fa
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FFT1DTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.builtin.relational.tvf.fft;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+
+public class FFT1DTest {
+
+ @Test
+ public void testDoubleComplexForwardPowerOfTwoLength() {
+ double[] values = {1.0, 0.5, -2.0, 1.0, 0.0, -1.5, 3.0, 2.0};
+ double[] expected = directDft(values);
+
+ new DoubleFFT_1D(4).complexForward(values);
+
+ assertArrayEquals(expected, values, 1e-9);
+ }
+
+ @Test
+ public void testDoubleComplexForwardNonPowerOfTwoLength() {
+ double[] values = {1.0, 0.0, 2.0, -0.5, -1.0, 1.5, 0.0, 0.25, 3.0, -2.0};
+ double[] expected = directDft(values);
+
+ new DoubleFFT_1D(5).complexForward(values);
+
+ assertArrayEquals(expected, values, 1e-9);
+ }
+
+ @Test
+ public void testFloatComplexForwardPowerOfTwoLength() {
+ float[] values = {1.0f, 0.5f, -2.0f, 1.0f, 0.0f, -1.5f, 3.0f, 2.0f};
+ float[] expected = directDft(values);
+
+ new FloatFFT_1D(4).complexForward(values);
+
+ assertArrayEquals(expected, values, 1e-5f);
+ }
+
+ @Test
+ public void testFloatComplexForwardNonPowerOfTwoLength() {
+ float[] values = {1.0f, 0.0f, 2.0f, -0.5f, -1.0f, 1.5f, 0.0f, 0.25f, 3.0f,
-2.0f};
+ float[] expected = directDft(values);
+
+ new FloatFFT_1D(5).complexForward(values);
+
+ assertArrayEquals(expected, values, 1e-5f);
+ }
+
+ private static double[] directDft(double[] values) {
+ int length = values.length / 2;
+ double[] result = new double[values.length];
+ for (int frequencyIndex = 0; frequencyIndex < length; frequencyIndex++) {
+ double real = 0.0;
+ double imaginary = 0.0;
+ for (int timeIndex = 0; timeIndex < length; timeIndex++) {
+ double angle = -2.0 * Math.PI * frequencyIndex * timeIndex / length;
+ double cos = Math.cos(angle);
+ double sin = Math.sin(angle);
+ double inputReal = values[2 * timeIndex];
+ double inputImaginary = values[2 * timeIndex + 1];
+ real += inputReal * cos - inputImaginary * sin;
+ imaginary += inputReal * sin + inputImaginary * cos;
+ }
+ result[2 * frequencyIndex] = real;
+ result[2 * frequencyIndex + 1] = imaginary;
+ }
+ return result;
+ }
+
+ private static float[] directDft(float[] values) {
+ int length = values.length / 2;
+ float[] result = new float[values.length];
+ for (int frequencyIndex = 0; frequencyIndex < length; frequencyIndex++) {
+ double real = 0.0;
+ double imaginary = 0.0;
+ for (int timeIndex = 0; timeIndex < length; timeIndex++) {
+ double angle = -2.0 * Math.PI * frequencyIndex * timeIndex / length;
+ double cos = Math.cos(angle);
+ double sin = Math.sin(angle);
+ double inputReal = values[2 * timeIndex];
+ double inputImaginary = values[2 * timeIndex + 1];
+ real += inputReal * cos - inputImaginary * sin;
+ imaginary += inputReal * sin + inputImaginary * cos;
+ }
+ result[2 * frequencyIndex] = (float) real;
+ result[2 * frequencyIndex + 1] = (float) imaginary;
+ }
+ return result;
+ }
+}
