From 2faa54bd5e7ebea4f75744ed7379dddafb244b83 Mon Sep 17 00:00:00 2001 From: DaZuiZui Date: Mon, 22 Jun 2026 15:50:03 +0800 Subject: [PATCH 1/9] Implement FFT table function --- .../it/db/it/IoTDBFFTTableFunctionIT.java | 223 ++++++ .../analyzer/StatementAnalyzer.java | 67 +- .../relational/planner/RelationPlanner.java | 3 + .../analyzer/TableFunctionTest.java | 136 ++++ iotdb-core/node-commons/pom.xml | 4 + .../function/TableBuiltinTableFunction.java | 4 + .../relational/tvf/FFTTableFunction.java | 731 ++++++++++++++++++ .../relational/tvf/FFTTableFunctionTest.java | 246 ++++++ 8 files changed, 1413 insertions(+), 1 deletion(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBFFTTableFunctionIT.java create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunction.java create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunctionTest.java 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 0000000000000..8eb95c9022578 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBFFTTableFunctionIT.java @@ -0,0 +1,223 @@ +/* + * 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.Statement; + +import static org.apache.iotdb.db.it.utils.TestUtils.tableAssertTestFail; +import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest; + +@RunWith(IoTDBTestRunner.class) +@Category({TableLocalStandaloneIT.class, TableClusterIT.class}) +public class IoTDBFFTTableFunctionIT { + + private static final String DATABASE_NAME = "test_fft"; + 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)", + "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" + }; + String[] retArray = + new String[] { + "d1,0,0.0,1.0,0.0,10.0,0.0,", + "d1,1,0.25,1.0,0.0,-2.0,2.0,", + "d1,2,-0.5,1.0,0.0,-2.0,0.0,", + "d1,3,-0.25,1.0,0.0,-2.0,-2.0,", + "d2,0,0.0,2.0,0.0,10.0,0.0,", + "d2,1,0.25,2.0,0.0,2.0,-2.0,", + "d2,2,-0.5,2.0,0.0,2.0,0.0,", + "d2,3,-0.25,2.0,0.0,2.0,2.0," + }; + + tableResultSetEqualTest( + "SELECT * FROM FFT(DATA => signal PARTITION BY device_id ORDER BY time, " + + "SAMPLE_INTERVAL => 1s, N => 4) ORDER BY device_id, frequency_index", + expectedHeader, + retArray, + DATABASE_NAME); + } + + @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 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 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 => irregular PARTITION BY device_id ORDER BY time)", + "701: FFT requires evenly spaced input time values when SAMPLE_INTERVAL is not specified.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM FFT(DATA => irregular PARTITION BY device_id ORDER BY time, SAMPLE_INTERVAL => 1s)", + "701: FFT input time interval must match the specified SAMPLE_INTERVAL.", + 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); + } +} 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 5a24cd4f09a94..106915579757a 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.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; @@ -5165,7 +5166,7 @@ public Scope visitTableFunctionInvocation(TableFunctionInvocation node, Optional 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)) @@ -5297,9 +5298,73 @@ private ArgumentsAnalysis analyzeArguments( } } 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 arguments, + List parameterSpecifications, + ImmutableMap.Builder passedArguments) { + if (!TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName)) { + return; + } + + Optional sampleIntervalArgument = + findOptionalTableFunctionArgument( + arguments, parameterSpecifications, FFTTableFunction.SAMPLE_INTERVAL_PARAMETER_NAME); + if (sampleIntervalArgument.isPresent() + && !(sampleIntervalArgument.get().getValue() instanceof TimeDurationLiteral)) { + throw new SemanticException( + "The SAMPLE_INTERVAL argument of FFT must be a duration literal."); + } + + Optional nArgument = + findOptionalTableFunctionArgument( + arguments, parameterSpecifications, FFTTableFunction.N_PARAMETER_NAME); + if (nArgument.isPresent() && nArgument.get().getValue() instanceof TimeDurationLiteral) { + throw new SemanticException("The N argument of FFT must be a 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 arguments, + List parameterSpecifications) { + Optional dataArgument = + findOptionalTableFunctionArgument( + arguments, parameterSpecifications, FFTTableFunction.DATA_PARAMETER_NAME); + if (!dataArgument.isPresent() + || !(dataArgument.get().getValue() instanceof TableFunctionTableArgument)) { + return; + } + + Optional 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( + "The ORDER BY clause of the DATA argument must sort the time column in ascending order."); + } + } + } + private void tryAppendM4ModeArgument( String functionName, List 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 bf14e37d6f3f9..4cce197f89e0b 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 @@ -1564,6 +1564,9 @@ public RelationPlan visitTableFunctionInvocation(TableFunctionInvocation node, V } 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/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 9d09ff057b30a..c6a057d6fa090 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 @@ 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,141 @@ public void testForecastFunctionAbnormal() { } } + @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.builder() + .put("time", "time") + .put("tag1_0", "tag1") + .put("s1", "s1") + .put("s2", "s2") + .put("s3", "s3") + .buildOrThrow()); + + Consumer 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", "s1,s2,s3") + .build()); + + assertPlan( + logicalQueryPlan, anyTree(tableFunctionProcessor(tableFunctionMatcher, 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.builder() + .put("time", "time") + .put("s1", "s1") + .put("s2", "s2") + .put("s3", "s3") + .buildOrThrow()); + + Consumer 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", "s1,s2,s3") + .build()); + + assertPlan( + logicalQueryPlan, anyTree(tableFunctionProcessor(tableFunctionMatcher, sort(tableScan)))); + } + + @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."); + 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/pom.xml b/iotdb-core/node-commons/pom.xml index f0a1afcb8221c..204ffc1cf851b 100644 --- a/iotdb-core/node-commons/pom.xml +++ b/iotdb-core/node-commons/pom.xml @@ -147,6 +147,10 @@ cglib cglib + + com.github.wendykierp + JTransforms + com.google.errorprone error_prone_annotations 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 eb969a2c6ae34..decaf8dc4669e 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.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 static TableFunction getBuiltinTableFunction(String functionName) { 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 0000000000000..e9d1011f337d0 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunction.java @@ -0,0 +1,731 @@ +/* + * 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.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 org.jtransforms.fft.DoubleFFT_1D; + +import java.util.ArrayList; +import java.util.Arrays; +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 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 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_DOUBLE_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 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 SUPPORTED_VALUE_TYPES = + new HashSet<>(Arrays.asList(Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE)); + + @Override + public List 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()); + } + + @Override + public TableFunctionAnalysis analyze(Map arguments) throws UDFException { + TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME); + if (tableArgument.getOrderBy().isEmpty()) { + throw new SemanticException("Table argument with set semantics requires an ORDER BY clause."); + } + + int timeColumnIndex = + findColumnIndex(tableArgument, TIME_COLUMN_NAME, Collections.singleton(Type.TIMESTAMP)); + validateOrderBy(tableArgument); + + List partitionIndexes = getPartitionIndexes(tableArgument); + Set excludedIndexes = new HashSet<>(partitionIndexes); + excludedIndexes.add(timeColumnIndex); + + List valueIndexes = new ArrayList<>(); + List valueNames = new ArrayList<>(); + List valueTypes = new ArrayList<>(); + List 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("FFT requires named numeric input 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("No numeric columns found for FFT calculation."); + } + + 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, joinStrings(valueNames)) + .build(); + + List 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 = splitStrings((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) { + if (tableArgument.getOrderBy().size() != 1 + || !tableArgument.getOrderBy().get(0).equalsIgnoreCase(TIME_COLUMN_NAME)) { + throw new SemanticException( + "The ORDER BY clause of the DATA argument must contain exactly the time column."); + } + } + + private static List getPartitionIndexes(TableArgument tableArgument) + throws UDFException { + List 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("FFT transform length N must not exceed %d.", MAX_TRANSFORM_LENGTH)); + } + long spectrumDoubleValues; + try { + spectrumDoubleValues = + Math.multiplyExact(Math.multiplyExact(transformLength, 2L), valueColumnCount); + } catch (ArithmeticException e) { + throw new SemanticException( + "FFT spectrum buffer is too large. Reduce N or the number of numeric columns."); + } + if (spectrumDoubleValues > MAX_SPECTRUM_DOUBLE_VALUES) { + throw new SemanticException( + "FFT spectrum buffer is too large. Reduce N or the number of numeric columns."); + } + } + + private static void validateNorm(String norm) { + if (!NORM_BACKWARD.equals(norm) && !NORM_FORWARD.equals(norm) && !NORM_ORTHO.equals(norm)) { + throw new SemanticException( + "Invalid NORM value for FFT. Supported values are backward, forward and ortho."); + } + } + + private static String joinTypes(List 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 joinStrings(List values) { + return String.join(",", values); + } + + private static String[] splitStrings(String value) { + if (value.isEmpty()) { + return new String[0]; + } + return value.split(","); + } + + 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("Unsupported FFT partition type: " + type); + } + } + + private enum NumericOperator { + INT32(Type.INT32) { + @Override + double read(Record record, int index) { + return record.getInt(index); + } + }, + INT64(Type.INT64) { + @Override + double read(Record record, int index) { + return record.getLong(index); + } + }, + FLOAT(Type.FLOAT) { + @Override + double read(Record record, int index) { + return record.getFloat(index); + } + }, + DOUBLE(Type.DOUBLE) { + @Override + double read(Record record, int index) { + return record.getDouble(index); + } + }; + + private final Type type; + + NumericOperator(Type type) { + this.type = type; + } + + abstract double read(Record record, int index); + + static NumericOperator fromType(Type type) { + for (NumericOperator numericOperator : values()) { + if (numericOperator.type == type) { + return numericOperator; + } + } + throw new IllegalArgumentException("Unsupported FFT 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 double read(Record record) { + return numericOperator.read(record, inputIndex); + } + } + + 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 rows = new ArrayList<>(); + private long previousTime; + private long inferredSampleInterval = UNSPECIFIED_SAMPLE_INTERVAL; + 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 properColumnBuilders, + ColumnBuilder passThroughIndexBuilder) { + long currentTime = input.getLong(0); + if (!initialized) { + capturePartitionValues(input); + initialized = true; + } else if (currentTime <= previousTime) { + throw new SemanticException( + "The time column of FFT input must be strictly ascending within each partition."); + } else { + validateSampleInterval(currentTime - previousTime); + } + previousTime = currentTime; + + double[] row = new double[valueColumns.length]; + for (int i = 0; i < valueColumns.length; i++) { + NumericColumn valueColumn = valueColumns[i]; + if (input.isNull(valueColumn.inputIndex)) { + throw new SemanticException( + String.format("FFT does not support null values in column [%s].", valueColumn.name)); + } + row[i] = valueColumn.read(input); + } + rows.add(row); + } + + @Override + public void finish( + List properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { + if (rows.isEmpty()) { + return; + } + + int transformLength = getTransformLength(); + double sampleIntervalSeconds = getSampleIntervalSeconds(); + double scaleFactor = getScaleFactor(transformLength); + double[][] spectra = new double[valueColumns.length][2 * transformLength]; + + int copiedRows = Math.min(rows.size(), transformLength); + DoubleFFT_1D fft = new DoubleFFT_1D(transformLength); + for (int columnIndex = 0; columnIndex < valueColumns.length; columnIndex++) { + double[] spectrum = spectra[columnIndex]; + for (int rowIndex = 0; rowIndex < copiedRows; rowIndex++) { + spectrum[2 * rowIndex] = rows.get(rowIndex)[columnIndex]; + } + fft.complexForward(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++) { + double[] spectrum = spectra[columnIndex]; + properColumnBuilders + .get(outputColumnIndex++) + .writeDouble(spectrum[2 * frequencyIndex] * scaleFactor); + properColumnBuilders + .get(outputColumnIndex++) + .writeDouble(spectrum[2 * frequencyIndex + 1] * 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 ? rows.size() : specifiedTransformLength; + validateTransformLength(transformLength, valueColumns.length); + return (int) transformLength; + } + + private double getSampleIntervalSeconds() { + long interval; + if (sampleIntervalSpecified) { + interval = sampleInterval; + } else { + if (rows.size() < 2) { + throw new SemanticException("FFT requires at least two rows to infer SAMPLE_INTERVAL."); + } + interval = inferredSampleInterval; + } + double intervalSeconds = + interval * TimestampPrecisionUtils.currPrecision.toNanos(1L) / 1_000_000_000.0; + if (intervalSeconds <= 0) { + throw new SemanticException("FFT SAMPLE_INTERVAL must be positive."); + } + return intervalSeconds; + } + + private void validateSampleInterval(long currentInterval) { + if (sampleIntervalSpecified) { + if (currentInterval != sampleInterval) { + throw new SemanticException( + "FFT input time interval must match the specified SAMPLE_INTERVAL."); + } + return; + } + + if (inferredSampleInterval == UNSPECIFIED_SAMPLE_INTERVAL) { + inferredSampleInterval = currentInterval; + } else if (currentInterval != inferredSampleInterval) { + throw new SemanticException( + "FFT requires evenly spaced input time values when SAMPLE_INTERVAL is not specified."); + } + } + + 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/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 0000000000000..456b024e5ff8b --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunctionTest.java @@ -0,0 +1,246 @@ +/* + * 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.udf.api.exception.UDFException; +import org.apache.iotdb.udf.api.relational.access.Record; +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.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.Map; +import java.util.Optional; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +public class FFTTableFunctionTest { + + private final FFTTableFunction function = new FFTTableFunction(); + + @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 testRejectsIrregularTimeWhenInferringSampleInterval() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(false); + processor.process(record(0L, 1.0), Collections.emptyList(), null); + processor.process(record(1L, 2.0), Collections.emptyList(), null); + + assertSemanticException( + () -> processor.process(record(3L, 3.0), Collections.emptyList(), null), + "FFT requires evenly spaced input time values when SAMPLE_INTERVAL is not specified."); + } + + @Test + public void testRejectsTimeGapDifferentFromExplicitSampleInterval() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(true); + processor.process(record(0L, 1.0), Collections.emptyList(), null); + + assertSemanticException( + () -> processor.process(record(2L, 2.0), Collections.emptyList(), null), + "FFT input time interval must match the specified SAMPLE_INTERVAL."); + } + + @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.finish(Collections.emptyList(), null), + "FFT transform length N must not exceed 65536."); + } + + private TableFunctionDataProcessor createProcessor(boolean sampleIntervalSpecified) + throws UDFException { + Map arguments = new HashMap<>(); + arguments.put( + FFTTableFunction.DATA_PARAMETER_NAME, + new TableArgument( + Arrays.asList(Optional.of("time"), Optional.of("value")), + Arrays.asList(Type.TIMESTAMP, Type.DOUBLE), + Collections.emptyList(), + Collections.singletonList("time"), + false)); + 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, -1L)); + arguments.put( + FFTTableFunction.NORM_PARAMETER_NAME, new ScalarArgument(Type.STRING, "backward")); + + return function + .getProcessorProvider(function.analyze(arguments).getTableFunctionHandle()) + .getDataProcessor(); + } + + private Record record(long time, double value) { + return new SimpleRecord(time, value); + } + + 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 double value; + + private SimpleRecord(long time, double 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) { + throw new UnsupportedOperationException(); + } + + @Override + public double getDouble(int columnIndex) { + if (columnIndex == 1) { + return value; + } + 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 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) { + return false; + } + + @Override + public int size() { + return 2; + } + } +} From a9ae6f466ac8d2462d4baa0cbdec5236dacd53f2 Mon Sep 17 00:00:00 2001 From: DaZuiZui Date: Tue, 23 Jun 2026 14:53:52 +0800 Subject: [PATCH 2/9] Add FFT table function coverage and notes --- RELEASE_NOTES.md | 1 + .../relational/tvf/FFTTableFunctionTest.java | 98 ++++++++++++++++++- 2 files changed, 95 insertions(+), 4 deletions(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 6637974a2bf4a..a843547a8cbf6 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -25,6 +25,7 @@ - Data Query: Added list display for available DataNode nodes - Data Query: Added a system table for statistics on query latency in the table model - Data Query: Python SessionDataset supported converting TsBlock to DataFrame and returning DataFrame in batches +- Data Query: Added the built-in FFT/fft table-valued function for the table model. SAMPLE_INTERVAL must be a positive duration literal when specified; N defaults to the partition row count and is capped at 65,536; null numeric values are rejected; timestamps must be strictly ascending and evenly spaced unless SAMPLE_INTERVAL is explicitly provided, in which case input gaps must match it. - Storage Management: Supported custom column names for the TIME column - Storage Management: Supported viewing the complete definition statement of created tables/views via SQL - System Management: Added a system table for DataNode node connection status in the table model 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 index 456b024e5ff8b..45a9dcd51d008 100644 --- 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 @@ -20,6 +20,7 @@ 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.argument.Argument; @@ -28,6 +29,10 @@ 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; @@ -36,6 +41,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Optional; @@ -44,8 +50,57 @@ 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 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 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 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 testRejectsDuplicateTime() throws UDFException { TableFunctionDataProcessor processor = createProcessor(false); @@ -111,6 +166,11 @@ public void testRejectsDefaultTransformLengthAboveLimit() throws UDFException { private TableFunctionDataProcessor createProcessor(boolean sampleIntervalSpecified) throws UDFException { + return createProcessor(sampleIntervalSpecified, -1L); + } + + private TableFunctionDataProcessor createProcessor( + boolean sampleIntervalSpecified, long transformLength) throws UDFException { Map arguments = new HashMap<>(); arguments.put( FFTTableFunction.DATA_PARAMETER_NAME, @@ -126,7 +186,8 @@ private TableFunctionDataProcessor createProcessor(boolean sampleIntervalSpecifi arguments.put( FFTTableFunction.SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME, new ScalarArgument(Type.BOOLEAN, sampleIntervalSpecified)); - arguments.put(FFTTableFunction.N_PARAMETER_NAME, new ScalarArgument(Type.INT64, -1L)); + arguments.put( + FFTTableFunction.N_PARAMETER_NAME, new ScalarArgument(Type.INT64, transformLength)); arguments.put( FFTTableFunction.NORM_PARAMETER_NAME, new ScalarArgument(Type.STRING, "backward")); @@ -139,6 +200,32 @@ private Record record(long time, double value) { return new SimpleRecord(time, value); } + private Record nullValueRecord(long time) { + return new SimpleRecord(time, null); + } + + private List 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) { + 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(); @@ -150,9 +237,9 @@ private void assertSemanticException(Runnable runnable, String message) { private static class SimpleRecord implements Record { private final long time; - private final double value; + private final Double value; - private SimpleRecord(long time, double value) { + private SimpleRecord(long time, Double value) { this.time = time; this.value = value; } @@ -177,7 +264,7 @@ public float getFloat(int columnIndex) { @Override public double getDouble(int columnIndex) { - if (columnIndex == 1) { + if (columnIndex == 1 && value != null) { return value; } throw new UnsupportedOperationException(); @@ -235,6 +322,9 @@ public Type getDataType(int columnIndex) { @Override public boolean isNull(int columnIndex) { + if (columnIndex == 1) { + return value == null; + } return false; } From 392ff7fd3faa12c9b907802516838bc322fe0574 Mon Sep 17 00:00:00 2001 From: DaZuiZui Date: Tue, 23 Jun 2026 15:50:29 +0800 Subject: [PATCH 3/9] Fix FFT buffering and distribution planning --- .../TableDistributedPlanGenerator.java | 6 ++++- .../analyzer/TableFunctionTest.java | 8 ++++++ .../relational/tvf/FFTTableFunction.java | 25 ++++++++++++++----- .../relational/tvf/FFTTableFunctionTest.java | 19 ++++++++++++-- 4 files changed, 49 insertions(+), 9 deletions(-) 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 1f55f9bc4972c..a0a3774489e51 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 @@ -1826,7 +1826,7 @@ public List visitTableFunctionProcessor( if (node.getChildren().isEmpty()) { return Collections.singletonList(node); } - boolean canSplitPushDown = node.isRowSemantic() || (node.getChild() instanceof GroupNode); + boolean canSplitPushDown = node.isRowSemantic() || isPartitionedGroup(node.getChild()); List childrenNodes = node.getChild().accept(this, context); if (childrenNodes.size() == 1) { node.setChild(childrenNodes.get(0)); @@ -1842,6 +1842,10 @@ public List visitTableFunctionProcessor( } } + private boolean isPartitionedGroup(PlanNode node) { + return node instanceof GroupNode && ((GroupNode) node).getPartitionKeyCount() > 0; + } + private void buildRegionNodeMap( AggregationTableScanNode originalAggTableScanNode, List> 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 c6a057d6fa090..347ec698fb01b 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 @@ -579,6 +579,14 @@ public void testFFTDefaultArguments() { 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 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 index e9d1011f337d0..bfce8501f01e6 100644 --- 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 @@ -557,6 +557,7 @@ private static class FFTDataProcessor implements TableFunctionDataProcessor { private final Object[] partitionValues; private final boolean[] partitionValueIsNull; private final List rows = new ArrayList<>(); + private long inputRowCount; private long previousTime; private long inferredSampleInterval = UNSPECIFIED_SAMPLE_INTERVAL; private boolean initialized; @@ -595,22 +596,34 @@ public void process( } previousTime = currentTime; - double[] row = new double[valueColumns.length]; + if (specifiedTransformLength == UNSPECIFIED_N && inputRowCount >= MAX_TRANSFORM_LENGTH) { + throw new SemanticException( + String.format("FFT transform length N must not exceed %d.", MAX_TRANSFORM_LENGTH)); + } + + boolean shouldCacheRow = + specifiedTransformLength == UNSPECIFIED_N || rows.size() < specifiedTransformLength; + double[] row = shouldCacheRow ? new double[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("FFT does not support null values in column [%s].", valueColumn.name)); } - row[i] = valueColumn.read(input); + if (shouldCacheRow) { + row[i] = valueColumn.read(input); + } + } + inputRowCount++; + if (shouldCacheRow) { + rows.add(row); } - rows.add(row); } @Override public void finish( List properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { - if (rows.isEmpty()) { + if (inputRowCount == 0) { return; } @@ -668,7 +681,7 @@ private void capturePartitionValues(Record input) { private int getTransformLength() { long transformLength = - specifiedTransformLength == UNSPECIFIED_N ? rows.size() : specifiedTransformLength; + specifiedTransformLength == UNSPECIFIED_N ? inputRowCount : specifiedTransformLength; validateTransformLength(transformLength, valueColumns.length); return (int) transformLength; } @@ -678,7 +691,7 @@ private double getSampleIntervalSeconds() { if (sampleIntervalSpecified) { interval = sampleInterval; } else { - if (rows.size() < 2) { + if (inputRowCount < 2) { throw new SemanticException("FFT requires at least two rows to infer SAMPLE_INTERVAL."); } interval = inferredSampleInterval; 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 index 45a9dcd51d008..a82a901f241f0 100644 --- 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 @@ -101,6 +101,21 @@ public void testRejectsInvalidRowsEvenWhenBeyondTruncatedN() throws UDFException "FFT does not support null values in column [value]."); } + @Test + public void testInfersSampleIntervalFromInputRowsBeyondSpecifiedN() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(false, 1L); + processor.process(record(0L, 1.0), Collections.emptyList(), null); + processor.process(record(2L, 2.0), Collections.emptyList(), null); + + List builders = createOutputBuilders(1); + processor.finish(builders, null); + + assertLongColumn(builders.get(0).build(), 0L); + assertDoubleColumn(builders.get(1).build(), 0.0); + assertDoubleColumn(builders.get(2).build(), 1.0); + assertDoubleColumn(builders.get(3).build(), 0.0); + } + @Test public void testRejectsDuplicateTime() throws UDFException { TableFunctionDataProcessor processor = createProcessor(false); @@ -155,12 +170,12 @@ public void testRejectsTimeGapDifferentFromExplicitSampleInterval() throws UDFEx @Test public void testRejectsDefaultTransformLengthAboveLimit() throws UDFException { TableFunctionDataProcessor processor = createProcessor(true); - for (long time = 0; time <= 65_536L; time++) { + for (long time = 0; time < 65_536L; time++) { processor.process(record(time, 1.0), Collections.emptyList(), null); } assertSemanticException( - () -> processor.finish(Collections.emptyList(), null), + () -> processor.process(record(65_536L, 1.0), Collections.emptyList(), null), "FFT transform length N must not exceed 65536."); } From 58ce71a50fecf2d5c43966af3d6ae6c76b08b0bc Mon Sep 17 00:00:00 2001 From: DaZuiZui Date: Mon, 6 Jul 2026 10:56:59 +0800 Subject: [PATCH 4/9] Vendor FFT implementation for table function --- LICENSE | 10 + iotdb-core/node-commons/pom.xml | 4 - .../builtin/relational/tvf/DoubleFft1d.java | 225 ++++++++++++++++++ .../relational/tvf/FFTTableFunction.java | 3 +- .../relational/tvf/FFTTableFunctionTest.java | 16 ++ 5 files changed, 252 insertions(+), 6 deletions(-) create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DoubleFft1d.java diff --git a/LICENSE b/LICENSE index 81000948cc4ef..d9ca6fa0d90c9 100644 --- a/LICENSE +++ b/LICENSE @@ -309,6 +309,16 @@ License: https://github.com/addthis/stream-lib/blob/master/LICENSE.txt -------------------------------------------------------------------------------- +The following file includes code modified from JTransforms project. + +./iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DoubleFft1d.java + +Copyright (c) 2007 onward, Piotr Wendykier +Project page: https://github.com/wendykierp/JTransforms +License: http://opensource.org/licenses/BSD-2-Clause + +-------------------------------------------------------------------------------- + The following files include code modified from Dropwizard Metrics project. ./iotdb-core/metrics/core/src/main/java/org/apache/iotdb/metrics/core/utils/IoTDBCachedGauge.java diff --git a/iotdb-core/node-commons/pom.xml b/iotdb-core/node-commons/pom.xml index 204ffc1cf851b..f0a1afcb8221c 100644 --- a/iotdb-core/node-commons/pom.xml +++ b/iotdb-core/node-commons/pom.xml @@ -147,10 +147,6 @@ cglib cglib - - com.github.wendykierp - JTransforms - com.google.errorprone error_prone_annotations diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DoubleFft1d.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DoubleFft1d.java new file mode 100644 index 0000000000000..947d8d2fbd1c6 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DoubleFft1d.java @@ -0,0 +1,225 @@ +/* + * 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; + +/** + * A small complex FFT implementation for {@link FFTTableFunction}. + * + *

This class follows the {@code complexForward(double[])} contract of JTransforms 3.1 {@code + * DoubleFFT_1D}, while keeping only the double-precision complex forward transform needed by + * IoTDB's built-in FFT table function. The implementation uses radix-2 Cooley-Tukey for power of + * two lengths and Bluestein convolution for other lengths. + * + *

Portions are derived from JTransforms 3.1, which carries the following notice: + * + * 

+ * JTransforms
+ * Copyright (c) 2007 onward, Piotr Wendykier
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice, this
+ *    list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+ * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */ +final class DoubleFft1d { + + private final int length; + private final int convolutionLength; + private final double[] chirpCos; + private final double[] chirpSin; + private final double[] kernelSpectrum; + + DoubleFft1d(int length) { + if (length < 1) { + throw new IllegalArgumentException("FFT length must be positive."); + } + this.length = length; + + if (isPowerOfTwo(length)) { + this.convolutionLength = 0; + this.chirpCos = null; + this.chirpSin = null; + this.kernelSpectrum = null; + return; + } + + this.convolutionLength = nextPowerOfTwo(2 * length - 1); + this.chirpCos = new double[length]; + this.chirpSin = new double[length]; + this.kernelSpectrum = new double[2 * convolutionLength]; + initializeBluesteinKernel(); + } + + void complexForward(double[] data) { + if (data.length < 2 * length) { + throw new IllegalArgumentException("The input array is too small for the FFT length."); + } + if (length == 1) { + return; + } + if (isPowerOfTwo(length)) { + transformRadix2(data, length, false); + } else { + transformBluestein(data); + } + } + + private void initializeBluesteinKernel() { + long period = 2L * length; + for (int i = 0; i < length; i++) { + double angle = Math.PI * ((long) i * i % period) / length; + double cos = Math.cos(angle); + double sin = Math.sin(angle); + chirpCos[i] = cos; + chirpSin[i] = sin; + kernelSpectrum[2 * i] = cos; + kernelSpectrum[2 * i + 1] = sin; + if (i > 0) { + int mirroredIndex = 2 * (convolutionLength - i); + kernelSpectrum[mirroredIndex] = cos; + kernelSpectrum[mirroredIndex + 1] = sin; + } + } + transformRadix2(kernelSpectrum, convolutionLength, false); + } + + private void transformBluestein(double[] data) { + double[] convolution = new double[2 * convolutionLength]; + for (int i = 0; i < length; i++) { + double dataReal = data[2 * i]; + double dataImag = data[2 * i + 1]; + double cos = chirpCos[i]; + double sin = chirpSin[i]; + convolution[2 * i] = dataReal * cos + dataImag * sin; + convolution[2 * i + 1] = dataImag * cos - dataReal * sin; + } + + transformRadix2(convolution, convolutionLength, false); + for (int i = 0; i < convolutionLength; i++) { + int index = 2 * i; + double real = convolution[index]; + double imag = convolution[index + 1]; + double kernelReal = kernelSpectrum[index]; + double kernelImag = kernelSpectrum[index + 1]; + convolution[index] = real * kernelReal - imag * kernelImag; + convolution[index + 1] = real * kernelImag + imag * kernelReal; + } + transformRadix2(convolution, convolutionLength, true); + + for (int i = 0; i < length; i++) { + double real = convolution[2 * i]; + double imag = convolution[2 * i + 1]; + double cos = chirpCos[i]; + double sin = chirpSin[i]; + data[2 * i] = real * cos + imag * sin; + data[2 * i + 1] = imag * cos - real * sin; + } + } + + private static void transformRadix2(double[] data, int complexLength, boolean inverse) { + bitReverse(data, complexLength); + + for (int size = 2; size <= complexLength; size <<= 1) { + int halfSize = size >>> 1; + double angle = (inverse ? 2.0 : -2.0) * Math.PI / size; + double stepReal = Math.cos(angle); + double stepImag = Math.sin(angle); + + for (int offset = 0; offset < complexLength; offset += size) { + double twiddleReal = 1.0; + double twiddleImag = 0.0; + for (int i = 0; i < halfSize; i++) { + int evenIndex = 2 * (offset + i); + int oddIndex = 2 * (offset + i + halfSize); + + double oddReal = data[oddIndex]; + double oddImag = data[oddIndex + 1]; + double transformedOddReal = oddReal * twiddleReal - oddImag * twiddleImag; + double transformedOddImag = oddReal * twiddleImag + oddImag * twiddleReal; + + double evenReal = data[evenIndex]; + double evenImag = data[evenIndex + 1]; + data[evenIndex] = evenReal + transformedOddReal; + data[evenIndex + 1] = evenImag + transformedOddImag; + data[oddIndex] = evenReal - transformedOddReal; + data[oddIndex + 1] = evenImag - transformedOddImag; + + double nextTwiddleReal = twiddleReal * stepReal - twiddleImag * stepImag; + twiddleImag = twiddleReal * stepImag + twiddleImag * stepReal; + twiddleReal = nextTwiddleReal; + } + } + } + + if (inverse) { + for (int i = 0; i < 2 * complexLength; i++) { + data[i] /= complexLength; + } + } + } + + private static void bitReverse(double[] data, int complexLength) { + int j = 0; + for (int i = 1; i < complexLength; i++) { + int bit = complexLength >>> 1; + while ((j & bit) != 0) { + j ^= bit; + bit >>>= 1; + } + j ^= bit; + if (i < j) { + swap(data, 2 * i, 2 * j); + swap(data, 2 * i + 1, 2 * j + 1); + } + } + } + + private static void swap(double[] data, int left, int right) { + double value = data[left]; + data[left] = data[right]; + data[right] = value; + } + + private static boolean isPowerOfTwo(int value) { + return (value & (value - 1)) == 0; + } + + private static int nextPowerOfTwo(int value) { + int highestOneBit = Integer.highestOneBit(value); + return highestOneBit == value ? value : highestOneBit << 1; + } +} 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 index bfce8501f01e6..f853312e8c397 100644 --- 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 @@ -40,7 +40,6 @@ import org.apache.tsfile.block.column.ColumnBuilder; import org.apache.tsfile.utils.Binary; -import org.jtransforms.fft.DoubleFFT_1D; import java.util.ArrayList; import java.util.Arrays; @@ -633,7 +632,7 @@ public void finish( double[][] spectra = new double[valueColumns.length][2 * transformLength]; int copiedRows = Math.min(rows.size(), transformLength); - DoubleFFT_1D fft = new DoubleFFT_1D(transformLength); + DoubleFft1d fft = new DoubleFft1d(transformLength); for (int columnIndex = 0; columnIndex < valueColumns.length; columnIndex++) { double[] spectrum = spectra[columnIndex]; for (int rowIndex = 0; rowIndex < copiedRows; rowIndex++) { 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 index a82a901f241f0..0b41e8451015d 100644 --- 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 @@ -90,6 +90,22 @@ public void testTruncatesInputRowsToSpecifiedN() throws UDFException { 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 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 testRejectsInvalidRowsEvenWhenBeyondTruncatedN() throws UDFException { TableFunctionDataProcessor processor = createProcessor(true, 2L); From 039c240071931124859207f0e82bcf1668ac22de Mon Sep 17 00:00:00 2001 From: DaZuiZui Date: Mon, 6 Jul 2026 17:30:06 +0800 Subject: [PATCH 5/9] Use JTransforms FFT for numeric inputs --- LICENSE | 10 - LICENSE-binary | 3 + .../it/db/it/IoTDBFFTTableFunctionIT.java | 65 +++-- iotdb-core/node-commons/pom.xml | 4 + .../builtin/relational/tvf/DoubleFft1d.java | 225 ------------------ .../relational/tvf/FFTTableFunction.java | 108 +++++++-- .../relational/tvf/FFTTableFunctionTest.java | 45 +++- 7 files changed, 178 insertions(+), 282 deletions(-) delete mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DoubleFft1d.java diff --git a/LICENSE b/LICENSE index d9ca6fa0d90c9..81000948cc4ef 100644 --- a/LICENSE +++ b/LICENSE @@ -309,16 +309,6 @@ License: https://github.com/addthis/stream-lib/blob/master/LICENSE.txt -------------------------------------------------------------------------------- -The following file includes code modified from JTransforms project. - -./iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DoubleFft1d.java - -Copyright (c) 2007 onward, Piotr Wendykier -Project page: https://github.com/wendykierp/JTransforms -License: http://opensource.org/licenses/BSD-2-Clause - --------------------------------------------------------------------------------- - The following files include code modified from Dropwizard Metrics project. ./iotdb-core/metrics/core/src/main/java/org/apache/iotdb/metrics/core/utils/IoTDBCachedGauge.java diff --git a/LICENSE-binary b/LICENSE-binary index d21627d28eeae..a9fd29be2f3d8 100644 --- a/LICENSE-binary +++ b/LICENSE-binary @@ -216,6 +216,7 @@ following license. See licenses/ for text of these licenses. Apache License 2.0 -------------------------------------- commons-cli:commons-cli:1.5.0 +org.apache.commons:commons-math3:3.5 com.google.code.gson:gson:2.13.1 com.google.guava.guava:32.1.2-jre com.fasterxml.jackson.core:jackson-annotations:2.16.2 @@ -273,6 +274,8 @@ org.jline:jline:3.26.2 BSD 2-Clause ------------ +com.github.wendykierp:JTransforms:3.1 +pl.edu.icm:JLargeArrays:1.5 com.github.luben:zstd-jni:1.5.6-3 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 index 8eb95c9022578..a4744f6575323 100644 --- 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 @@ -31,16 +31,22 @@ 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, @@ -104,24 +110,20 @@ public void testFFTWithPartitionAndMultipleColumns() { "speed_real", "speed_imag" }; - String[] retArray = - new String[] { - "d1,0,0.0,1.0,0.0,10.0,0.0,", - "d1,1,0.25,1.0,0.0,-2.0,2.0,", - "d1,2,-0.5,1.0,0.0,-2.0,0.0,", - "d1,3,-0.25,1.0,0.0,-2.0,-2.0,", - "d2,0,0.0,2.0,0.0,10.0,0.0,", - "d2,1,0.25,2.0,0.0,2.0,-2.0,", - "d2,2,-0.5,2.0,0.0,2.0,0.0,", - "d2,3,-0.25,2.0,0.0,2.0,2.0," - }; - - tableResultSetEqualTest( + 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, - retArray, - DATABASE_NAME); + 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 @@ -220,4 +222,37 @@ public void testFFTFailures() { "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/node-commons/pom.xml b/iotdb-core/node-commons/pom.xml index f0a1afcb8221c..204ffc1cf851b 100644 --- a/iotdb-core/node-commons/pom.xml +++ b/iotdb-core/node-commons/pom.xml @@ -147,6 +147,10 @@ cglib cglib + + com.github.wendykierp + JTransforms + com.google.errorprone error_prone_annotations diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DoubleFft1d.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DoubleFft1d.java deleted file mode 100644 index 947d8d2fbd1c6..0000000000000 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/DoubleFft1d.java +++ /dev/null @@ -1,225 +0,0 @@ -/* - * 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; - -/** - * A small complex FFT implementation for {@link FFTTableFunction}. - * - *

This class follows the {@code complexForward(double[])} contract of JTransforms 3.1 {@code - * DoubleFFT_1D}, while keeping only the double-precision complex forward transform needed by - * IoTDB's built-in FFT table function. The implementation uses radix-2 Cooley-Tukey for power of - * two lengths and Bluestein convolution for other lengths. - * - *

Portions are derived from JTransforms 3.1, which carries the following notice: - * - * 

- * JTransforms
- * Copyright (c) 2007 onward, Piotr Wendykier
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are met:
- *
- * 1. Redistributions of source code must retain the above copyright notice, this
- *    list of conditions and the following disclaimer.
- * 2. Redistributions in binary form must reproduce the above copyright notice,
- *    this list of conditions and the following disclaimer in the documentation
- *    and/or other materials provided with the distribution.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
- * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
- * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
- * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
- * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
- * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
- * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
- * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
- * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */ -final class DoubleFft1d { - - private final int length; - private final int convolutionLength; - private final double[] chirpCos; - private final double[] chirpSin; - private final double[] kernelSpectrum; - - DoubleFft1d(int length) { - if (length < 1) { - throw new IllegalArgumentException("FFT length must be positive."); - } - this.length = length; - - if (isPowerOfTwo(length)) { - this.convolutionLength = 0; - this.chirpCos = null; - this.chirpSin = null; - this.kernelSpectrum = null; - return; - } - - this.convolutionLength = nextPowerOfTwo(2 * length - 1); - this.chirpCos = new double[length]; - this.chirpSin = new double[length]; - this.kernelSpectrum = new double[2 * convolutionLength]; - initializeBluesteinKernel(); - } - - void complexForward(double[] data) { - if (data.length < 2 * length) { - throw new IllegalArgumentException("The input array is too small for the FFT length."); - } - if (length == 1) { - return; - } - if (isPowerOfTwo(length)) { - transformRadix2(data, length, false); - } else { - transformBluestein(data); - } - } - - private void initializeBluesteinKernel() { - long period = 2L * length; - for (int i = 0; i < length; i++) { - double angle = Math.PI * ((long) i * i % period) / length; - double cos = Math.cos(angle); - double sin = Math.sin(angle); - chirpCos[i] = cos; - chirpSin[i] = sin; - kernelSpectrum[2 * i] = cos; - kernelSpectrum[2 * i + 1] = sin; - if (i > 0) { - int mirroredIndex = 2 * (convolutionLength - i); - kernelSpectrum[mirroredIndex] = cos; - kernelSpectrum[mirroredIndex + 1] = sin; - } - } - transformRadix2(kernelSpectrum, convolutionLength, false); - } - - private void transformBluestein(double[] data) { - double[] convolution = new double[2 * convolutionLength]; - for (int i = 0; i < length; i++) { - double dataReal = data[2 * i]; - double dataImag = data[2 * i + 1]; - double cos = chirpCos[i]; - double sin = chirpSin[i]; - convolution[2 * i] = dataReal * cos + dataImag * sin; - convolution[2 * i + 1] = dataImag * cos - dataReal * sin; - } - - transformRadix2(convolution, convolutionLength, false); - for (int i = 0; i < convolutionLength; i++) { - int index = 2 * i; - double real = convolution[index]; - double imag = convolution[index + 1]; - double kernelReal = kernelSpectrum[index]; - double kernelImag = kernelSpectrum[index + 1]; - convolution[index] = real * kernelReal - imag * kernelImag; - convolution[index + 1] = real * kernelImag + imag * kernelReal; - } - transformRadix2(convolution, convolutionLength, true); - - for (int i = 0; i < length; i++) { - double real = convolution[2 * i]; - double imag = convolution[2 * i + 1]; - double cos = chirpCos[i]; - double sin = chirpSin[i]; - data[2 * i] = real * cos + imag * sin; - data[2 * i + 1] = imag * cos - real * sin; - } - } - - private static void transformRadix2(double[] data, int complexLength, boolean inverse) { - bitReverse(data, complexLength); - - for (int size = 2; size <= complexLength; size <<= 1) { - int halfSize = size >>> 1; - double angle = (inverse ? 2.0 : -2.0) * Math.PI / size; - double stepReal = Math.cos(angle); - double stepImag = Math.sin(angle); - - for (int offset = 0; offset < complexLength; offset += size) { - double twiddleReal = 1.0; - double twiddleImag = 0.0; - for (int i = 0; i < halfSize; i++) { - int evenIndex = 2 * (offset + i); - int oddIndex = 2 * (offset + i + halfSize); - - double oddReal = data[oddIndex]; - double oddImag = data[oddIndex + 1]; - double transformedOddReal = oddReal * twiddleReal - oddImag * twiddleImag; - double transformedOddImag = oddReal * twiddleImag + oddImag * twiddleReal; - - double evenReal = data[evenIndex]; - double evenImag = data[evenIndex + 1]; - data[evenIndex] = evenReal + transformedOddReal; - data[evenIndex + 1] = evenImag + transformedOddImag; - data[oddIndex] = evenReal - transformedOddReal; - data[oddIndex + 1] = evenImag - transformedOddImag; - - double nextTwiddleReal = twiddleReal * stepReal - twiddleImag * stepImag; - twiddleImag = twiddleReal * stepImag + twiddleImag * stepReal; - twiddleReal = nextTwiddleReal; - } - } - } - - if (inverse) { - for (int i = 0; i < 2 * complexLength; i++) { - data[i] /= complexLength; - } - } - } - - private static void bitReverse(double[] data, int complexLength) { - int j = 0; - for (int i = 1; i < complexLength; i++) { - int bit = complexLength >>> 1; - while ((j & bit) != 0) { - j ^= bit; - bit >>>= 1; - } - j ^= bit; - if (i < j) { - swap(data, 2 * i, 2 * j); - swap(data, 2 * i + 1, 2 * j + 1); - } - } - } - - private static void swap(double[] data, int left, int right) { - double value = data[left]; - data[left] = data[right]; - data[right] = value; - } - - private static boolean isPowerOfTwo(int value) { - return (value & (value - 1)) == 0; - } - - private static int nextPowerOfTwo(int value) { - int highestOneBit = Integer.highestOneBit(value); - return highestOneBit == value ? value : highestOneBit << 1; - } -} 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 index f853312e8c397..7ab781ba287f2 100644 --- 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 @@ -40,6 +40,8 @@ import org.apache.tsfile.block.column.ColumnBuilder; import org.apache.tsfile.utils.Binary; +import org.jtransforms.fft.DoubleFFT_1D; +import org.jtransforms.fft.FloatFFT_1D; import java.util.ArrayList; import java.util.Arrays; @@ -71,7 +73,7 @@ public class FFTTableFunction implements TableFunction { 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_DOUBLE_VALUES = 16_777_216L; + 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"; @@ -263,15 +265,15 @@ public static void validateTransformLength(long transformLength, int valueColumn throw new SemanticException( String.format("FFT transform length N must not exceed %d.", MAX_TRANSFORM_LENGTH)); } - long spectrumDoubleValues; + long spectrumValues; try { - spectrumDoubleValues = + spectrumValues = Math.multiplyExact(Math.multiplyExact(transformLength, 2L), valueColumnCount); } catch (ArithmeticException e) { throw new SemanticException( "FFT spectrum buffer is too large. Reduce N or the number of numeric columns."); } - if (spectrumDoubleValues > MAX_SPECTRUM_DOUBLE_VALUES) { + if (spectrumValues > MAX_SPECTRUM_VALUES) { throw new SemanticException( "FFT spectrum buffer is too large. Reduce N or the number of numeric columns."); } @@ -469,38 +471,44 @@ static ValueOperator fromType(Type type) { } private enum NumericOperator { - INT32(Type.INT32) { + INT32(Type.INT32, false) { @Override - double read(Record record, int index) { + Number read(Record record, int index) { return record.getInt(index); } }, - INT64(Type.INT64) { + INT64(Type.INT64, false) { @Override - double read(Record record, int index) { + Number read(Record record, int index) { return record.getLong(index); } }, - FLOAT(Type.FLOAT) { + FLOAT(Type.FLOAT, true) { @Override - double read(Record record, int index) { + Number read(Record record, int index) { return record.getFloat(index); } }, - DOUBLE(Type.DOUBLE) { + DOUBLE(Type.DOUBLE, false) { @Override - double read(Record record, int index) { + Number read(Record record, int index) { return record.getDouble(index); } }; private final Type type; + private final boolean floatFft; - NumericOperator(Type type) { + NumericOperator(Type type, boolean floatFft) { this.type = type; + this.floatFft = floatFft; } - abstract double read(Record record, int index); + abstract Number read(Record record, int index); + + boolean usesFloatFft() { + return floatFft; + } static NumericOperator fromType(Type type) { for (NumericOperator numericOperator : values()) { @@ -541,9 +549,41 @@ private NumericColumn(int inputIndex, String name, NumericOperator numericOperat this.numericOperator = numericOperator; } - private double read(Record record) { + 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 { @@ -555,7 +595,7 @@ private static class FFTDataProcessor implements TableFunctionDataProcessor { private final NumericColumn[] valueColumns; private final Object[] partitionValues; private final boolean[] partitionValueIsNull; - private final List rows = new ArrayList<>(); + private final List rows = new ArrayList<>(); private long inputRowCount; private long previousTime; private long inferredSampleInterval = UNSPECIFIED_SAMPLE_INTERVAL; @@ -602,7 +642,7 @@ public void process( boolean shouldCacheRow = specifiedTransformLength == UNSPECIFIED_N || rows.size() < specifiedTransformLength; - double[] row = shouldCacheRow ? new double[valueColumns.length] : null; + 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)) { @@ -629,16 +669,33 @@ public void finish( int transformLength = getTransformLength(); double sampleIntervalSeconds = getSampleIntervalSeconds(); double scaleFactor = getScaleFactor(transformLength); - double[][] spectra = new double[valueColumns.length][2 * transformLength]; + Spectrum[] spectra = new Spectrum[valueColumns.length]; int copiedRows = Math.min(rows.size(), transformLength); - DoubleFft1d fft = new DoubleFft1d(transformLength); + DoubleFFT_1D doubleFft = null; + FloatFFT_1D floatFft = null; for (int columnIndex = 0; columnIndex < valueColumns.length; columnIndex++) { - double[] spectrum = spectra[columnIndex]; - for (int rowIndex = 0; rowIndex < copiedRows; rowIndex++) { - spectrum[2 * rowIndex] = rows.get(rowIndex)[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); } - fft.complexForward(spectrum); } for (int frequencyIndex = 0; frequencyIndex < transformLength; frequencyIndex++) { @@ -657,13 +714,12 @@ public void finish( .writeDouble( calculateFrequency(frequencyIndex, transformLength, sampleIntervalSeconds)); for (int columnIndex = 0; columnIndex < valueColumns.length; columnIndex++) { - double[] spectrum = spectra[columnIndex]; properColumnBuilders .get(outputColumnIndex++) - .writeDouble(spectrum[2 * frequencyIndex] * scaleFactor); + .writeDouble(spectra[columnIndex].real(frequencyIndex, scaleFactor)); properColumnBuilders .get(outputColumnIndex++) - .writeDouble(spectrum[2 * frequencyIndex + 1] * scaleFactor); + .writeDouble(spectra[columnIndex].imaginary(frequencyIndex, scaleFactor)); } } } 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 index 0b41e8451015d..999d47015b4cc 100644 --- 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 @@ -106,6 +106,23 @@ public void testSupportsNonPowerOfTwoTransformLength() throws UDFException { 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 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); @@ -202,12 +219,17 @@ private TableFunctionDataProcessor createProcessor(boolean sampleIntervalSpecifi 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 arguments = new HashMap<>(); arguments.put( FFTTableFunction.DATA_PARAMETER_NAME, new TableArgument( Arrays.asList(Optional.of("time"), Optional.of("value")), - Arrays.asList(Type.TIMESTAMP, Type.DOUBLE), + Arrays.asList(Type.TIMESTAMP, valueType), Collections.emptyList(), Collections.singletonList("time"), false)); @@ -231,6 +253,10 @@ 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); } @@ -251,9 +277,13 @@ private void assertLongColumn(Column column, long... expected) { } 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); + assertEquals(expected[i], column.getDouble(i), delta); } } @@ -268,9 +298,9 @@ private void assertSemanticException(Runnable runnable, String message) { private static class SimpleRecord implements Record { private final long time; - private final Double value; + private final Number value; - private SimpleRecord(long time, Double value) { + private SimpleRecord(long time, Number value) { this.time = time; this.value = value; } @@ -290,13 +320,16 @@ public long getLong(int columnIndex) { @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 != null) { - return value; + if (columnIndex == 1 && value instanceof Double) { + return value.doubleValue(); } throw new UnsupportedOperationException(); } From a61970f55d6831f0b796f44d70c346e470a32f13 Mon Sep 17 00:00:00 2001 From: DaZuiZui Date: Tue, 7 Jul 2026 11:21:21 +0800 Subject: [PATCH 6/9] Support custom FFT time column --- .../it/db/it/IoTDBFFTTableFunctionIT.java | 17 ++++++ .../analyzer/TableFunctionTest.java | 16 +++++- .../relational/tvf/FFTTableFunction.java | 20 ++++--- .../relational/tvf/FFTTableFunctionTest.java | 53 +++++++++++++++++++ 4 files changed, 99 insertions(+), 7 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 index a4744f6575323..33dec77a653a7 100644 --- 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 @@ -71,6 +71,11 @@ public class IoTDBFFTTableFunctionIT { "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" }; @@ -142,6 +147,18 @@ public void testFFTDefaultNAndInferredSampleInterval() { 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( 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 347ec698fb01b..252c41e53156f 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 @@ -589,6 +589,20 @@ public void testFFTDefaultArguments() { 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 testFFTRejectsInvalidArguments() { assertAnalyzeFails( @@ -599,7 +613,7 @@ public void testFFTRejectsInvalidArguments() { "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."); + "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."); 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 index 7ab781ba287f2..d8978ee28526e 100644 --- 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 @@ -58,13 +58,14 @@ 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 TIME_COLUMN_NAME = "time"; + 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"; @@ -97,6 +98,11 @@ public class FFTTableFunction implements TableFunction { public List getArgumentsSpecifications() { return Arrays.asList( TableParameterSpecification.builder().name(DATA_PARAMETER_NAME).setSemantics().build(), + ScalarParameterSpecification.builder() + .name(TIMECOL_PARAMETER_NAME) + .type(Type.STRING) + .defaultValue(DEFAULT_TIME_COLUMN_NAME) + .build(), ScalarParameterSpecification.builder() .name(SAMPLE_INTERVAL_PARAMETER_NAME) .type(Type.INT64) @@ -123,9 +129,11 @@ public TableFunctionAnalysis analyze(Map arguments) throws UDF throw new SemanticException("Table argument with set semantics requires an ORDER BY clause."); } + String timeColumn = + (String) ((ScalarArgument) arguments.get(TIMECOL_PARAMETER_NAME)).getValue(); int timeColumnIndex = - findColumnIndex(tableArgument, TIME_COLUMN_NAME, Collections.singleton(Type.TIMESTAMP)); - validateOrderBy(tableArgument); + findColumnIndex(tableArgument, timeColumn, Collections.singleton(Type.TIMESTAMP)); + validateOrderBy(tableArgument, timeColumn); List partitionIndexes = getPartitionIndexes(tableArgument); Set excludedIndexes = new HashSet<>(partitionIndexes); @@ -240,11 +248,11 @@ public TableFunctionDataProcessor getDataProcessor() { }; } - private static void validateOrderBy(TableArgument tableArgument) { + private static void validateOrderBy(TableArgument tableArgument, String timeColumn) { if (tableArgument.getOrderBy().size() != 1 - || !tableArgument.getOrderBy().get(0).equalsIgnoreCase(TIME_COLUMN_NAME)) { + || !tableArgument.getOrderBy().get(0).equalsIgnoreCase(timeColumn)) { throw new SemanticException( - "The ORDER BY clause of the DATA argument must contain exactly the time column."); + "The ORDER BY clause of the DATA argument must contain exactly the time column specified by the TIMECOL argument."); } } 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 index 999d47015b4cc..d85acd488c645 100644 --- 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 @@ -23,6 +23,7 @@ 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.argument.Argument; import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument; import org.apache.iotdb.udf.api.relational.table.argument.TableArgument; @@ -212,6 +213,34 @@ public void testRejectsDefaultTransformLengthAboveLimit() throws UDFException { "FFT transform length N must not exceed 65536."); } + @Test + public void testAnalyzeUsesSpecifiedTimeColumn() throws UDFException { + Map 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 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); @@ -233,6 +262,7 @@ private TableFunctionDataProcessor createProcessor( 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)); @@ -249,6 +279,29 @@ private TableFunctionDataProcessor createProcessor( .getDataProcessor(); } + private Map createArguments(String timeColumn, String orderByColumn) { + Map arguments = new HashMap<>(); + arguments.put( + FFTTableFunction.DATA_PARAMETER_NAME, + new TableArgument( + Arrays.asList(Optional.of(timeColumn), Optional.of("value")), + 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); } From 32baae8ecf45b9b8a12d14acb068c0ef81959a66 Mon Sep 17 00:00:00 2001 From: DaZuiZui Date: Tue, 7 Jul 2026 15:22:16 +0800 Subject: [PATCH 7/9] Preserve FFT positional argument order --- .../plan/relational/analyzer/TableFunctionTest.java | 8 ++++++++ .../udf/builtin/relational/tvf/FFTTableFunction.java | 10 +++++----- 2 files changed, 13 insertions(+), 5 deletions(-) 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 252c41e53156f..180d7eb33dc53 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 @@ -603,6 +603,14 @@ public void testFFTWithSpecifiedTimeColumn() { 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( 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 index d8978ee28526e..a8d6fd238d0b8 100644 --- 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 @@ -98,11 +98,6 @@ public class FFTTableFunction implements TableFunction { public List getArgumentsSpecifications() { return Arrays.asList( TableParameterSpecification.builder().name(DATA_PARAMETER_NAME).setSemantics().build(), - ScalarParameterSpecification.builder() - .name(TIMECOL_PARAMETER_NAME) - .type(Type.STRING) - .defaultValue(DEFAULT_TIME_COLUMN_NAME) - .build(), ScalarParameterSpecification.builder() .name(SAMPLE_INTERVAL_PARAMETER_NAME) .type(Type.INT64) @@ -119,6 +114,11 @@ public List getArgumentsSpecifications() { .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()); } From ad200416abbd809e42f8ce7148ed9873351d7b35 Mon Sep 17 00:00:00 2001 From: DaZuiZui Date: Tue, 7 Jul 2026 16:48:55 +0800 Subject: [PATCH 8/9] Align FFT sample interval handling --- RELEASE_NOTES.md | 2 +- .../it/db/it/IoTDBFFTTableFunctionIT.java | 34 +++++++++---- .../relational/tvf/FFTTableFunction.java | 26 ++-------- .../relational/tvf/FFTTableFunctionTest.java | 48 ++++++++++++------- 4 files changed, 63 insertions(+), 47 deletions(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index a843547a8cbf6..3f3116ab1dd9b 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -25,7 +25,7 @@ - Data Query: Added list display for available DataNode nodes - Data Query: Added a system table for statistics on query latency in the table model - Data Query: Python SessionDataset supported converting TsBlock to DataFrame and returning DataFrame in batches -- Data Query: Added the built-in FFT/fft table-valued function for the table model. SAMPLE_INTERVAL must be a positive duration literal when specified; N defaults to the partition row count and is capped at 65,536; null numeric values are rejected; timestamps must be strictly ascending and evenly spaced unless SAMPLE_INTERVAL is explicitly provided, in which case input gaps must match it. +- Data Query: Added the built-in FFT/fft table-valued function for the table model. SAMPLE_INTERVAL must be a positive duration literal when specified; when omitted, the interval is inferred from the partition time range; N defaults to the partition row count and is capped at 65,536; null numeric values are rejected; timestamps must be strictly ascending. - Storage Management: Supported custom column names for the TIME column - Storage Management: Supported viewing the complete definition statement of created tables/views via SQL - System Management: Added a system table for DataNode node connection status in the table model 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 index 33dec77a653a7..874f5bf9076ee 100644 --- 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 @@ -208,6 +208,32 @@ public void testFFTNorm() { 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( @@ -226,14 +252,6 @@ public void testFFTFailures() { "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 => irregular PARTITION BY device_id ORDER BY time)", - "701: FFT requires evenly spaced input time values when SAMPLE_INTERVAL is not specified.", - DATABASE_NAME); - tableAssertTestFail( - "SELECT * FROM FFT(DATA => irregular PARTITION BY device_id ORDER BY time, SAMPLE_INTERVAL => 1s)", - "701: FFT input time interval must match the specified SAMPLE_INTERVAL.", - 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.", 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 index a8d6fd238d0b8..8d01fd6751f21 100644 --- 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 @@ -605,8 +605,8 @@ private static class FFTDataProcessor implements TableFunctionDataProcessor { private final boolean[] partitionValueIsNull; private final List rows = new ArrayList<>(); private long inputRowCount; + private long firstTime; private long previousTime; - private long inferredSampleInterval = UNSPECIFIED_SAMPLE_INTERVAL; private boolean initialized; private FFTDataProcessor( @@ -634,12 +634,11 @@ public void process( long currentTime = input.getLong(0); if (!initialized) { capturePartitionValues(input); + firstTime = currentTime; initialized = true; } else if (currentTime <= previousTime) { throw new SemanticException( "The time column of FFT input must be strictly ascending within each partition."); - } else { - validateSampleInterval(currentTime - previousTime); } previousTime = currentTime; @@ -750,14 +749,14 @@ private int getTransformLength() { } private double getSampleIntervalSeconds() { - long interval; + double interval; if (sampleIntervalSpecified) { interval = sampleInterval; } else { if (inputRowCount < 2) { throw new SemanticException("FFT requires at least two rows to infer SAMPLE_INTERVAL."); } - interval = inferredSampleInterval; + interval = (double) (previousTime - firstTime) / (inputRowCount - 1); } double intervalSeconds = interval * TimestampPrecisionUtils.currPrecision.toNanos(1L) / 1_000_000_000.0; @@ -767,23 +766,6 @@ private double getSampleIntervalSeconds() { return intervalSeconds; } - private void validateSampleInterval(long currentInterval) { - if (sampleIntervalSpecified) { - if (currentInterval != sampleInterval) { - throw new SemanticException( - "FFT input time interval must match the specified SAMPLE_INTERVAL."); - } - return; - } - - if (inferredSampleInterval == UNSPECIFIED_SAMPLE_INTERVAL) { - inferredSampleInterval = currentInterval; - } else if (currentInterval != inferredSampleInterval) { - throw new SemanticException( - "FFT requires evenly spaced input time values when SAMPLE_INTERVAL is not specified."); - } - } - private double getScaleFactor(int transformLength) { if (NORM_FORWARD.equals(norm)) { return 1.0 / transformLength; 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 index d85acd488c645..68df8e1da3e7e 100644 --- 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 @@ -136,18 +136,21 @@ public void testRejectsInvalidRowsEvenWhenBeyondTruncatedN() throws UDFException } @Test - public void testInfersSampleIntervalFromInputRowsBeyondSpecifiedN() throws UDFException { - TableFunctionDataProcessor processor = createProcessor(false, 1L); + public void testInfersSampleIntervalFromFullInputRangeBeyondSpecifiedN() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(false, 2L); processor.process(record(0L, 1.0), Collections.emptyList(), null); - processor.process(record(2L, 2.0), Collections.emptyList(), null); + processor.process(record(1L, 2.0), Collections.emptyList(), null); + processor.process(record(3L, 3.0), Collections.emptyList(), null); - List builders = createOutputBuilders(1); + List builders = createOutputBuilders(2); processor.finish(builders, null); - assertLongColumn(builders.get(0).build(), 0L); - assertDoubleColumn(builders.get(1).build(), 0.0); - assertDoubleColumn(builders.get(2).build(), 1.0); - assertDoubleColumn(builders.get(3).build(), 0.0); + 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 @@ -181,24 +184,37 @@ public void testRejectsSingleRowWithoutSampleInterval() throws UDFException { } @Test - public void testRejectsIrregularTimeWhenInferringSampleInterval() throws UDFException { + 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); - assertSemanticException( - () -> processor.process(record(3L, 3.0), Collections.emptyList(), null), - "FFT requires evenly spaced input time values when SAMPLE_INTERVAL is not specified."); + List 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 testRejectsTimeGapDifferentFromExplicitSampleInterval() throws UDFException { + 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); - assertSemanticException( - () -> processor.process(record(2L, 2.0), Collections.emptyList(), null), - "FFT input time interval must match the specified SAMPLE_INTERVAL."); + List 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 From 725686817a084f776c88da66a42e6d8c43e5dd78 Mon Sep 17 00:00:00 2001 From: DaZuiZui Date: Wed, 8 Jul 2026 13:34:31 +0800 Subject: [PATCH 9/9] Guard default FFT spectrum buffer growth --- .../udf/builtin/relational/tvf/FFTTableFunction.java | 5 ++--- .../udf/builtin/relational/tvf/FFTTableFunctionTest.java | 7 +++++++ 2 files changed, 9 insertions(+), 3 deletions(-) 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 index 8d01fd6751f21..d35f2e93ff7a9 100644 --- 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 @@ -642,9 +642,8 @@ public void process( } previousTime = currentTime; - if (specifiedTransformLength == UNSPECIFIED_N && inputRowCount >= MAX_TRANSFORM_LENGTH) { - throw new SemanticException( - String.format("FFT transform length N must not exceed %d.", MAX_TRANSFORM_LENGTH)); + if (specifiedTransformLength == UNSPECIFIED_N) { + validateTransformLength(inputRowCount + 1, valueColumns.length); } boolean shouldCacheRow = 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 index 68df8e1da3e7e..313a66e5235fb 100644 --- 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 @@ -229,6 +229,13 @@ public void testRejectsDefaultTransformLengthAboveLimit() throws UDFException { "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 arguments = createArguments("event_time", "event_time");