[#17939] Implement FFT table function for table model

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
 
 

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; 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

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBFFTTableFunctionIT.java

@@ -0,0 +1,293 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.it.db.it;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableClusterIT;
+import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.tableAssertTestFail;
+import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+public class IoTDBFFTTableFunctionIT {
+
+  private static final String DATABASE_NAME = "test_fft";
+  private static final double DELTA = 1e-9;
+  private static final String[] SQLS =
+      new String[] {
+        "CREATE DATABASE " + DATABASE_NAME,
+        "USE " + DATABASE_NAME,
+        "CREATE TABLE signal(device_id STRING TAG, temperature DOUBLE FIELD, speed INT32 FIELD, note STRING FIELD)",
+        "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (0, 'd1', 1.0, 1, 'ok')",
+        "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (1000, 'd1', 0.0, 2, 'ok')",
+        "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (2000, 'd1', 0.0, 3, 'ok')",
+        "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (3000, 'd1', 0.0, 4, 'ok')",
+        "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (0, 'd2', 2.0, 4, 'ok')",
+        "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (1000, 'd2', 0.0, 3, 'ok')",
+        "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (2000, 'd2', 0.0, 2, 'ok')",
+        "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (3000, 'd2', 0.0, 1, 'ok')",
+        "CREATE TABLE single_row(device_id STRING TAG, value DOUBLE FIELD)",
+        "INSERT INTO single_row(time, device_id, value) VALUES (0, 'd1', 1.0)",
+        "CREATE TABLE no_numeric(device_id STRING TAG, note STRING FIELD)",
+        "INSERT INTO no_numeric(time, device_id, note) VALUES (0, 'd1', 'ok')",
+        "CREATE TABLE with_null(device_id STRING TAG, value DOUBLE FIELD)",
+        "INSERT INTO with_null(time, device_id, value) VALUES (0, 'd1', 1.0)",
+        "INSERT INTO with_null(time, device_id, value) VALUES (1000, 'd1', null)",
+        "CREATE TABLE irregular(device_id STRING TAG, value DOUBLE FIELD)",
+        "INSERT INTO irregular(time, device_id, value) VALUES (0, 'd1', 1.0)",
+        "INSERT INTO irregular(time, device_id, value) VALUES (1000, 'd1', 2.0)",
+        "INSERT INTO irregular(time, device_id, value) VALUES (2500, 'd1', 3.0)",
+        "CREATE TABLE custom_time_signal(event_time TIMESTAMP TIME, value DOUBLE FIELD)",
+        "INSERT INTO custom_time_signal(event_time, value) VALUES (0, 1.0)",
+        "INSERT INTO custom_time_signal(event_time, value) VALUES (1000, 0.0)",
+        "INSERT INTO custom_time_signal(event_time, value) VALUES (2000, 0.0)",
+        "INSERT INTO custom_time_signal(event_time, value) VALUES (3000, 0.0)",
+        "FLUSH"
+      };
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    EnvFactory.getEnv().initClusterEnvironment();
+    insertData();
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  private static void insertData() {
+    String currentSql = null;
+    try (Connection connection = EnvFactory.getEnv().getTableConnection();
+        Statement statement = connection.createStatement()) {
+      for (String sql : SQLS) {
+        currentSql = sql;
+        statement.execute(sql);
+      }
+    } catch (Exception e) {
+      throw new AssertionError("insertData failed while executing [" + currentSql + "].", e);
+    }
+  }
+
+  @Test
+  public void testFFTWithPartitionAndMultipleColumns() {
+    String[] expectedHeader =
+        new String[] {
+          "device_id",
+          "frequency_index",
+          "frequency",
+          "temperature_real",
+          "temperature_imag",
+          "speed_real",
+          "speed_imag"
+        };
+    assertFftRows(
+        "SELECT * FROM FFT(DATA => signal PARTITION BY device_id ORDER BY time, "
+            + "SAMPLE_INTERVAL => 1s, N => 4) ORDER BY device_id, frequency_index",
+        expectedHeader,
+        new Object[][] {
+          {"d1", 0L, 0.0, 1.0, 0.0, 10.0, 0.0},
+          {"d1", 1L, 0.25, 1.0, 0.0, -2.0, 2.0},
+          {"d1", 2L, -0.5, 1.0, 0.0, -2.0, 0.0},
+          {"d1", 3L, -0.25, 1.0, 0.0, -2.0, -2.0},
+          {"d2", 0L, 0.0, 2.0, 0.0, 10.0, 0.0},
+          {"d2", 1L, 0.25, 2.0, 0.0, 2.0, -2.0},
+          {"d2", 2L, -0.5, 2.0, 0.0, 2.0, 0.0},
+          {"d2", 3L, -0.25, 2.0, 0.0, 2.0, 2.0}
+        });
+  }
+
+  @Test
+  public void testFFTDefaultNAndInferredSampleInterval() {
+    String[] expectedHeader =
+        new String[] {"frequency_index", "frequency", "temperature_real", "temperature_imag"};
+    String[] retArray =
+        new String[] {"0,0.0,1.0,0.0,", "1,0.25,1.0,0.0,", "2,-0.5,1.0,0.0,", "3,-0.25,1.0,0.0,"};
+
+    tableResultSetEqualTest(
+        "SELECT frequency_index, frequency, temperature_real, temperature_imag "
+            + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE device_id='d1') "
+            + "ORDER BY time) ORDER BY frequency_index",
+        expectedHeader,
+        retArray,
+        DATABASE_NAME);
+  }
+
+  @Test
+  public void testFFTWithSpecifiedTimeColumn() {
+    tableResultSetEqualTest(
+        "SELECT frequency_index, frequency, value_real, value_imag "
+            + "FROM FFT(DATA => custom_time_signal ORDER BY event_time, "
+            + "TIMECOL => 'event_time', SAMPLE_INTERVAL => 1s, N => 4) "
+            + "ORDER BY frequency_index",
+        new String[] {"frequency_index", "frequency", "value_real", "value_imag"},
+        new String[] {"0,0.0,1.0,0.0,", "1,0.25,1.0,0.0,", "2,-0.5,1.0,0.0,", "3,-0.25,1.0,0.0,"},
+        DATABASE_NAME);
+  }
+
+  @Test
+  public void testFFTTruncateAndZeroPad() {
+    tableResultSetEqualTest(
+        "SELECT frequency_index, frequency, speed_real, speed_imag "
+            + "FROM FFT(DATA => (SELECT time, speed FROM signal WHERE device_id='d1') "
+            + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 2) ORDER BY frequency_index",
+        new String[] {"frequency_index", "frequency", "speed_real", "speed_imag"},
+        new String[] {"0,0.0,3.0,0.0,", "1,-0.5,-1.0,0.0,"},
+        DATABASE_NAME);
+
+    tableResultSetEqualTest(
+        "SELECT frequency_index, frequency, temperature_real, temperature_imag "
+            + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE device_id='d1') "
+            + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 8) ORDER BY frequency_index",
+        new String[] {"frequency_index", "frequency", "temperature_real", "temperature_imag"},
+        new String[] {
+          "0,0.0,1.0,0.0,",
+          "1,0.125,1.0,0.0,",
+          "2,0.25,1.0,0.0,",
+          "3,0.375,1.0,0.0,",
+          "4,-0.5,1.0,0.0,",
+          "5,-0.375,1.0,0.0,",
+          "6,-0.25,1.0,0.0,",
+          "7,-0.125,1.0,0.0,"
+        },
+        DATABASE_NAME);
+  }
+
+  @Test
+  public void testFFTNorm() {
+    tableResultSetEqualTest(
+        "SELECT frequency_index, temperature_real, temperature_imag "
+            + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE device_id='d1') "
+            + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 4, NORM => 'forward') "
+            + "ORDER BY frequency_index",
+        new String[] {"frequency_index", "temperature_real", "temperature_imag"},
+        new String[] {"0,0.25,0.0,", "1,0.25,0.0,", "2,0.25,0.0,", "3,0.25,0.0,"},
+        DATABASE_NAME);
+
+    tableResultSetEqualTest(
+        "SELECT frequency_index, temperature_real, temperature_imag "
+            + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE device_id='d1') "
+            + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 4, NORM => 'ortho') "
+            + "ORDER BY frequency_index",
+        new String[] {"frequency_index", "temperature_real", "temperature_imag"},
+        new String[] {"0,0.5,0.0,", "1,0.5,0.0,", "2,0.5,0.0,", "3,0.5,0.0,"},
+        DATABASE_NAME);
+  }
+
+  @Test
+  public void testFFTIrregularTimeUsesInferredOrExplicitSampleInterval() {
+    String[] expectedHeader =
+        new String[] {"device_id", "frequency_index", "frequency", "value_real", "value_imag"};
+
+    assertFftRows(
+        "SELECT * FROM FFT(DATA => irregular PARTITION BY device_id ORDER BY time) "
+            + "ORDER BY frequency_index",
+        expectedHeader,
+        new Object[][] {
+          {"d1", 0L, 0.0, 6.0, 0.0},
+          {"d1", 1L, 0.26666666666666666, -1.5, 0.8660254037844386},
+          {"d1", 2L, -0.26666666666666666, -1.5, -0.8660254037844386}
+        });
+
+    assertFftRows(
+        "SELECT * FROM FFT(DATA => irregular PARTITION BY device_id ORDER BY time, "
+            + "SAMPLE_INTERVAL => 1s) ORDER BY frequency_index",
+        expectedHeader,
+        new Object[][] {
+          {"d1", 0L, 0.0, 6.0, 0.0},
+          {"d1", 1L, 0.3333333333333333, -1.5, 0.8660254037844386},
+          {"d1", 2L, -0.3333333333333333, -1.5, -0.8660254037844386}
+        });
+  }
+
+  @Test
+  public void testFFTFailures() {
+    tableAssertTestFail(
+        "SELECT * FROM FFT(DATA => signal PARTITION BY device_id, SAMPLE_INTERVAL => 1s)",
+        "701: Table argument with set semantics requires an ORDER BY clause.",
+        DATABASE_NAME);
+    tableAssertTestFail(
+        "SELECT * FROM FFT(DATA => single_row PARTITION BY device_id ORDER BY time)",
+        "701: FFT requires at least two rows to infer SAMPLE_INTERVAL.",
+        DATABASE_NAME);
+    tableAssertTestFail(
+        "SELECT * FROM FFT(DATA => no_numeric PARTITION BY device_id ORDER BY time, SAMPLE_INTERVAL => 1s)",
+        "701: No numeric columns found for FFT calculation.",
+        DATABASE_NAME);
+    tableAssertTestFail(
+        "SELECT * FROM FFT(DATA => with_null PARTITION BY device_id ORDER BY time, SAMPLE_INTERVAL => 1s)",
+        "701: FFT does not support null values in column [value].",
+        DATABASE_NAME);
+    tableAssertTestFail(
+        "SELECT * FROM FFT(DATA => signal PARTITION BY device_id ORDER BY time, N => 65537)",
+        "701: FFT transform length N must not exceed 65536.",
+        DATABASE_NAME);
+  }
+
+  private static void assertFftRows(String sql, String[] expectedHeader, Object[][] expectedRows) {
+    try (Connection connection = EnvFactory.getEnv().getTableConnection();
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE " + DATABASE_NAME);
+      try (ResultSet resultSet = statement.executeQuery(sql)) {
+        assertHeader(resultSet.getMetaData(), expectedHeader);
+
+        int rowIndex = 0;
+        while (resultSet.next()) {
+          Object[] expectedRow = expectedRows[rowIndex];
+          assertEquals(expectedRow[0], resultSet.getString(1));
+          assertEquals(expectedRow[1], resultSet.getLong(2));
+          for (int columnIndex = 2; columnIndex < expectedRow.length; columnIndex++) {
+            assertEquals(
+                (double) expectedRow[columnIndex], resultSet.getDouble(columnIndex + 1), DELTA);
+          }
+          rowIndex++;
+        }
+        assertEquals(expectedRows.length, rowIndex);
+      }
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+  }
+
+  private static void assertHeader(ResultSetMetaData metaData, String[] expectedHeader)
+      throws SQLException {
+    assertEquals(expectedHeader.length, metaData.getColumnCount());
+    for (int i = 1; i <= metaData.getColumnCount(); i++) {
+      assertEquals(expectedHeader[i - 1], metaData.getColumnName(i));
+    }
+  }
+}