diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 6637974a2bf4a..3f3116ab1dd9b 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; 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 new file mode 100644 index 0000000000000..874f5bf9076ee --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBFFTTableFunctionIT.java @@ -0,0 +1,293 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.relational.it.db.it; + +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.TableClusterIT; +import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Statement; + +import static org.apache.iotdb.db.it.utils.TestUtils.tableAssertTestFail; +import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +@RunWith(IoTDBTestRunner.class) +@Category({TableLocalStandaloneIT.class, TableClusterIT.class}) +public class IoTDBFFTTableFunctionIT { + + private static final String DATABASE_NAME = "test_fft"; + private static final double DELTA = 1e-9; + private static final String[] SQLS = + new String[] { + "CREATE DATABASE " + DATABASE_NAME, + "USE " + DATABASE_NAME, + "CREATE TABLE signal(device_id STRING TAG, temperature DOUBLE FIELD, speed INT32 FIELD, note STRING FIELD)", + "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (0, 'd1', 1.0, 1, 'ok')", + "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (1000, 'd1', 0.0, 2, 'ok')", + "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (2000, 'd1', 0.0, 3, 'ok')", + "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (3000, 'd1', 0.0, 4, 'ok')", + "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (0, 'd2', 2.0, 4, 'ok')", + "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (1000, 'd2', 0.0, 3, 'ok')", + "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (2000, 'd2', 0.0, 2, 'ok')", + "INSERT INTO signal(time, device_id, temperature, speed, note) VALUES (3000, 'd2', 0.0, 1, 'ok')", + "CREATE TABLE single_row(device_id STRING TAG, value DOUBLE FIELD)", + "INSERT INTO single_row(time, device_id, value) VALUES (0, 'd1', 1.0)", + "CREATE TABLE no_numeric(device_id STRING TAG, note STRING FIELD)", + "INSERT INTO no_numeric(time, device_id, note) VALUES (0, 'd1', 'ok')", + "CREATE TABLE with_null(device_id STRING TAG, value DOUBLE FIELD)", + "INSERT INTO with_null(time, device_id, value) VALUES (0, 'd1', 1.0)", + "INSERT INTO with_null(time, device_id, value) VALUES (1000, 'd1', null)", + "CREATE TABLE irregular(device_id STRING TAG, value DOUBLE FIELD)", + "INSERT INTO irregular(time, device_id, value) VALUES (0, 'd1', 1.0)", + "INSERT INTO irregular(time, device_id, value) VALUES (1000, 'd1', 2.0)", + "INSERT INTO irregular(time, device_id, value) VALUES (2500, 'd1', 3.0)", + "CREATE TABLE custom_time_signal(event_time TIMESTAMP TIME, value DOUBLE FIELD)", + "INSERT INTO custom_time_signal(event_time, value) VALUES (0, 1.0)", + "INSERT INTO custom_time_signal(event_time, value) VALUES (1000, 0.0)", + "INSERT INTO custom_time_signal(event_time, value) VALUES (2000, 0.0)", + "INSERT INTO custom_time_signal(event_time, value) VALUES (3000, 0.0)", + "FLUSH" + }; + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv().initClusterEnvironment(); + insertData(); + } + + @AfterClass + public static void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + private static void insertData() { + String currentSql = null; + try (Connection connection = EnvFactory.getEnv().getTableConnection(); + Statement statement = connection.createStatement()) { + for (String sql : SQLS) { + currentSql = sql; + statement.execute(sql); + } + } catch (Exception e) { + throw new AssertionError("insertData failed while executing [" + currentSql + "].", e); + } + } + + @Test + public void testFFTWithPartitionAndMultipleColumns() { + String[] expectedHeader = + new String[] { + "device_id", + "frequency_index", + "frequency", + "temperature_real", + "temperature_imag", + "speed_real", + "speed_imag" + }; + assertFftRows( + "SELECT * FROM FFT(DATA => signal PARTITION BY device_id ORDER BY time, " + + "SAMPLE_INTERVAL => 1s, N => 4) ORDER BY device_id, frequency_index", + expectedHeader, + new Object[][] { + {"d1", 0L, 0.0, 1.0, 0.0, 10.0, 0.0}, + {"d1", 1L, 0.25, 1.0, 0.0, -2.0, 2.0}, + {"d1", 2L, -0.5, 1.0, 0.0, -2.0, 0.0}, + {"d1", 3L, -0.25, 1.0, 0.0, -2.0, -2.0}, + {"d2", 0L, 0.0, 2.0, 0.0, 10.0, 0.0}, + {"d2", 1L, 0.25, 2.0, 0.0, 2.0, -2.0}, + {"d2", 2L, -0.5, 2.0, 0.0, 2.0, 0.0}, + {"d2", 3L, -0.25, 2.0, 0.0, 2.0, 2.0} + }); + } + + @Test + public void testFFTDefaultNAndInferredSampleInterval() { + String[] expectedHeader = + new String[] {"frequency_index", "frequency", "temperature_real", "temperature_imag"}; + String[] retArray = + new String[] {"0,0.0,1.0,0.0,", "1,0.25,1.0,0.0,", "2,-0.5,1.0,0.0,", "3,-0.25,1.0,0.0,"}; + + tableResultSetEqualTest( + "SELECT frequency_index, frequency, temperature_real, temperature_imag " + + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE device_id='d1') " + + "ORDER BY time) ORDER BY frequency_index", + expectedHeader, + retArray, + DATABASE_NAME); + } + + @Test + public void testFFTWithSpecifiedTimeColumn() { + tableResultSetEqualTest( + "SELECT frequency_index, frequency, value_real, value_imag " + + "FROM FFT(DATA => custom_time_signal ORDER BY event_time, " + + "TIMECOL => 'event_time', SAMPLE_INTERVAL => 1s, N => 4) " + + "ORDER BY frequency_index", + new String[] {"frequency_index", "frequency", "value_real", "value_imag"}, + new String[] {"0,0.0,1.0,0.0,", "1,0.25,1.0,0.0,", "2,-0.5,1.0,0.0,", "3,-0.25,1.0,0.0,"}, + DATABASE_NAME); + } + + @Test + public void testFFTTruncateAndZeroPad() { + tableResultSetEqualTest( + "SELECT frequency_index, frequency, speed_real, speed_imag " + + "FROM FFT(DATA => (SELECT time, speed FROM signal WHERE device_id='d1') " + + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 2) ORDER BY frequency_index", + new String[] {"frequency_index", "frequency", "speed_real", "speed_imag"}, + new String[] {"0,0.0,3.0,0.0,", "1,-0.5,-1.0,0.0,"}, + DATABASE_NAME); + + tableResultSetEqualTest( + "SELECT frequency_index, frequency, temperature_real, temperature_imag " + + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE device_id='d1') " + + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 8) ORDER BY frequency_index", + new String[] {"frequency_index", "frequency", "temperature_real", "temperature_imag"}, + new String[] { + "0,0.0,1.0,0.0,", + "1,0.125,1.0,0.0,", + "2,0.25,1.0,0.0,", + "3,0.375,1.0,0.0,", + "4,-0.5,1.0,0.0,", + "5,-0.375,1.0,0.0,", + "6,-0.25,1.0,0.0,", + "7,-0.125,1.0,0.0," + }, + DATABASE_NAME); + } + + @Test + public void testFFTNorm() { + tableResultSetEqualTest( + "SELECT frequency_index, temperature_real, temperature_imag " + + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE device_id='d1') " + + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 4, NORM => 'forward') " + + "ORDER BY frequency_index", + new String[] {"frequency_index", "temperature_real", "temperature_imag"}, + new String[] {"0,0.25,0.0,", "1,0.25,0.0,", "2,0.25,0.0,", "3,0.25,0.0,"}, + DATABASE_NAME); + + tableResultSetEqualTest( + "SELECT frequency_index, temperature_real, temperature_imag " + + "FROM FFT(DATA => (SELECT time, temperature FROM signal WHERE device_id='d1') " + + "ORDER BY time, SAMPLE_INTERVAL => 1s, N => 4, NORM => 'ortho') " + + "ORDER BY frequency_index", + new String[] {"frequency_index", "temperature_real", "temperature_imag"}, + new String[] {"0,0.5,0.0,", "1,0.5,0.0,", "2,0.5,0.0,", "3,0.5,0.0,"}, + DATABASE_NAME); + } + + @Test + public void testFFTIrregularTimeUsesInferredOrExplicitSampleInterval() { + String[] expectedHeader = + new String[] {"device_id", "frequency_index", "frequency", "value_real", "value_imag"}; + + assertFftRows( + "SELECT * FROM FFT(DATA => irregular PARTITION BY device_id ORDER BY time) " + + "ORDER BY frequency_index", + expectedHeader, + new Object[][] { + {"d1", 0L, 0.0, 6.0, 0.0}, + {"d1", 1L, 0.26666666666666666, -1.5, 0.8660254037844386}, + {"d1", 2L, -0.26666666666666666, -1.5, -0.8660254037844386} + }); + + assertFftRows( + "SELECT * FROM FFT(DATA => irregular PARTITION BY device_id ORDER BY time, " + + "SAMPLE_INTERVAL => 1s) ORDER BY frequency_index", + expectedHeader, + new Object[][] { + {"d1", 0L, 0.0, 6.0, 0.0}, + {"d1", 1L, 0.3333333333333333, -1.5, 0.8660254037844386}, + {"d1", 2L, -0.3333333333333333, -1.5, -0.8660254037844386} + }); + } + + @Test + public void testFFTFailures() { + tableAssertTestFail( + "SELECT * FROM FFT(DATA => signal PARTITION BY device_id, SAMPLE_INTERVAL => 1s)", + "701: Table argument with set semantics requires an ORDER BY clause.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM FFT(DATA => single_row PARTITION BY device_id ORDER BY time)", + "701: FFT requires at least two rows to infer SAMPLE_INTERVAL.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM FFT(DATA => no_numeric PARTITION BY device_id ORDER BY time, SAMPLE_INTERVAL => 1s)", + "701: No numeric columns found for FFT calculation.", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM FFT(DATA => with_null PARTITION BY device_id ORDER BY time, SAMPLE_INTERVAL => 1s)", + "701: FFT does not support null values in column [value].", + DATABASE_NAME); + tableAssertTestFail( + "SELECT * FROM FFT(DATA => signal PARTITION BY device_id ORDER BY time, N => 65537)", + "701: FFT transform length N must not exceed 65536.", + DATABASE_NAME); + } + + private static void assertFftRows(String sql, String[] expectedHeader, Object[][] expectedRows) { + try (Connection connection = EnvFactory.getEnv().getTableConnection(); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE_NAME); + try (ResultSet resultSet = statement.executeQuery(sql)) { + assertHeader(resultSet.getMetaData(), expectedHeader); + + int rowIndex = 0; + while (resultSet.next()) { + Object[] expectedRow = expectedRows[rowIndex]; + assertEquals(expectedRow[0], resultSet.getString(1)); + assertEquals(expectedRow[1], resultSet.getLong(2)); + for (int columnIndex = 2; columnIndex < expectedRow.length; columnIndex++) { + assertEquals( + (double) expectedRow[columnIndex], resultSet.getDouble(columnIndex + 1), DELTA); + } + rowIndex++; + } + assertEquals(expectedRows.length, rowIndex); + } + } catch (SQLException e) { + fail(e.getMessage()); + } + } + + private static void assertHeader(ResultSetMetaData metaData, String[] expectedHeader) + throws SQLException { + assertEquals(expectedHeader.length, metaData.getColumnCount()); + for (int i = 1; i <= metaData.getColumnCount(); i++) { + assertEquals(expectedHeader[i - 1], metaData.getColumnName(i)); + } + } +} diff --git a/iotdb-client/client-cpp/src/include/Session.h b/iotdb-client/client-cpp/src/include/Session.h index bd9d77f3bba5f..a0910584e5776 100644 --- a/iotdb-client/client-cpp/src/include/Session.h +++ b/iotdb-client/client-cpp/src/include/Session.h @@ -232,6 +232,14 @@ class Tablet { } void addTimestamp(size_t rowIndex, int64_t timestamp) { + if (rowIndex >= static_cast(maxRowNumber)) { + char tmpStr[100]; + snprintf(tmpStr, sizeof(tmpStr), + "Tablet::addTimestamp(), rowIndex >= maxRowNumber. rowIndex=%ld, " + "maxRowNumber=%ld.", + (long)rowIndex, (long)maxRowNumber); + throw std::out_of_range(tmpStr); + } timestamps[rowIndex] = timestamp; rowSize = max(rowSize, rowIndex + 1); } @@ -248,10 +256,18 @@ class Tablet { throw std::out_of_range(tmpStr); } - if (rowIndex >= rowSize) { + if (rowIndex >= static_cast(maxRowNumber)) { + char tmpStr[100]; + snprintf(tmpStr, sizeof(tmpStr), + "Tablet::addValue(), rowIndex >= maxRowNumber. rowIndex=%ld, maxRowNumber=%ld.", + (long)rowIndex, (long)maxRowNumber); + throw std::out_of_range(tmpStr); + } + + if (rowIndex >= static_cast(rowSize)) { char tmpStr[100]; snprintf(tmpStr, sizeof(tmpStr), - "Tablet::addValue(), rowIndex >= rowSize. rowIndex=%ld, rowSize.size()=%ld.", + "Tablet::addValue(), rowIndex >= rowSize. rowIndex=%ld, rowSize=%ld.", (long)rowIndex, (long)rowSize); throw std::out_of_range(tmpStr); } diff --git a/iotdb-client/client-cpp/src/rpc/SessionImpl.h b/iotdb-client/client-cpp/src/rpc/SessionImpl.h index 4d14c99f38146..9fc3d91729342 100644 --- a/iotdb-client/client-cpp/src/rpc/SessionImpl.h +++ b/iotdb-client/client-cpp/src/rpc/SessionImpl.h @@ -60,6 +60,13 @@ class Session::Impl { std::shared_ptr nodesSupplier_; std::shared_ptr defaultSessionConnection_; + std::shared_ptr getDefaultSessionConnection() { + if (isClosed_ || !defaultSessionConnection_) { + throw IoTDBConnectionException("Session is not open, please invoke Session.open() first"); + } + return defaultSessionConnection_; + } + TEndPoint defaultEndPoint_; struct TEndPointHash { @@ -168,7 +175,7 @@ void Session::Impl::insertByGroup( if (endPointToSessionConnection.size() > 1) { removeBrokenSessionConnection(connection); try { - insertConsumer(defaultSessionConnection_, req); + insertConsumer(getDefaultSessionConnection(), req); } catch (const RedirectException&) { } } else { @@ -216,7 +223,7 @@ void Session::Impl::insertOnce( if (endPointToSessionConnection.size() > 1) { removeBrokenSessionConnection(connection); try { - insertConsumer(defaultSessionConnection_, req); + insertConsumer(getDefaultSessionConnection(), req); } catch (const RedirectException&) { } } else { diff --git a/iotdb-client/client-cpp/src/session/Session.cpp b/iotdb-client/client-cpp/src/session/Session.cpp index 3af4de3bcce38..7dd0a7813ed50 100644 --- a/iotdb-client/client-cpp/src/session/Session.cpp +++ b/iotdb-client/client-cpp/src/session/Session.cpp @@ -1067,6 +1067,31 @@ void Session::close() { if (impl_->isClosed_) { return; } + try { + if (impl_->enableRedirection_) { + for (auto& entry : impl_->endPointToSessionConnection) { + if (entry.second) { + try { + entry.second->close(); + } catch (const exception& e) { + log_debug(e.what()); + } + } + } + impl_->endPointToSessionConnection.clear(); + } + if (impl_->defaultSessionConnection_) { + try { + impl_->defaultSessionConnection_->close(); + } catch (const exception& e) { + log_debug(e.what()); + } + impl_->defaultSessionConnection_.reset(); + } + } catch (...) { + impl_->isClosed_ = true; + throw; + } impl_->isClosed_ = true; } @@ -1087,7 +1112,7 @@ void Session::insertRecord(const string& deviceId, int64_t time, const vectordeviceIdToEndpoint.find(deviceId) != impl_->deviceIdToEndpoint.end()) { impl_->deviceIdToEndpoint.erase(deviceId); try { - impl_->defaultSessionConnection_->insertStringRecord(req); + impl_->getDefaultSessionConnection()->insertStringRecord(req); } catch (RedirectException& e) { } } else { @@ -1116,7 +1141,7 @@ void Session::insertRecord(const string& deviceId, int64_t time, const vectordeviceIdToEndpoint.find(deviceId) != impl_->deviceIdToEndpoint.end()) { impl_->deviceIdToEndpoint.erase(deviceId); try { - impl_->defaultSessionConnection_->insertRecord(req); + impl_->getDefaultSessionConnection()->insertRecord(req); } catch (RedirectException& e) { } } else { @@ -1143,7 +1168,7 @@ void Session::insertAlignedRecord(const string& deviceId, int64_t time, impl_->deviceIdToEndpoint.find(deviceId) != impl_->deviceIdToEndpoint.end()) { impl_->deviceIdToEndpoint.erase(deviceId); try { - impl_->defaultSessionConnection_->insertStringRecord(req); + impl_->getDefaultSessionConnection()->insertStringRecord(req); } catch (RedirectException& e) { } } else { @@ -1173,7 +1198,7 @@ void Session::insertAlignedRecord(const string& deviceId, int64_t time, impl_->deviceIdToEndpoint.find(deviceId) != impl_->deviceIdToEndpoint.end()) { impl_->deviceIdToEndpoint.erase(deviceId); try { - impl_->defaultSessionConnection_->insertRecord(req); + impl_->getDefaultSessionConnection()->insertRecord(req); } catch (RedirectException& e) { } } else { @@ -1202,7 +1227,7 @@ void Session::insertRecords(const vector& deviceIds, const vectordefaultSessionConnection_->insertStringRecords(request); + impl_->getDefaultSessionConnection()->insertStringRecords(request); } catch (RedirectException& e) { } } @@ -1235,7 +1260,7 @@ void Session::insertRecords(const vector& deviceIds, const vectordefaultSessionConnection_->insertRecords(request); + impl_->getDefaultSessionConnection()->insertRecords(request); } catch (RedirectException& e) { } } @@ -1260,7 +1285,7 @@ void Session::insertAlignedRecords(const vector& deviceIds, const vector request.__set_valuesList(valuesList); request.__set_isAligned(true); try { - impl_->defaultSessionConnection_->insertStringRecords(request); + impl_->getDefaultSessionConnection()->insertStringRecords(request); } catch (RedirectException& e) { } } @@ -1293,7 +1318,7 @@ void Session::insertAlignedRecords(const vector& deviceIds, const vector request.__set_valuesList(bufferList); request.__set_isAligned(false); try { - impl_->defaultSessionConnection_->insertRecords(request); + impl_->getDefaultSessionConnection()->insertRecords(request); } catch (RedirectException& e) { } } @@ -1345,7 +1370,7 @@ void Session::insertRecordsOfOneDevice(const string& deviceId, vector& impl_->deviceIdToEndpoint.find(deviceId) != impl_->deviceIdToEndpoint.end()) { impl_->deviceIdToEndpoint.erase(deviceId); try { - impl_->defaultSessionConnection_->insertRecordsOfOneDevice(request); + impl_->getDefaultSessionConnection()->insertRecordsOfOneDevice(request); } catch (RedirectException& e) { } } else { @@ -1400,7 +1425,7 @@ void Session::insertAlignedRecordsOfOneDevice(const string& deviceId, vectordeviceIdToEndpoint.find(deviceId) != impl_->deviceIdToEndpoint.end()) { impl_->deviceIdToEndpoint.erase(deviceId); try { - impl_->defaultSessionConnection_->insertRecordsOfOneDevice(request); + impl_->getDefaultSessionConnection()->insertRecordsOfOneDevice(request); } catch (RedirectException& e) { } } else { @@ -1452,7 +1477,7 @@ void Session::Impl::insertTablet(TSInsertTabletReq request) { if (enableRedirection_ && deviceIdToEndpoint.find(deviceId) != deviceIdToEndpoint.end()) { deviceIdToEndpoint.erase(deviceId); try { - defaultSessionConnection_->insertTablet(request); + getDefaultSessionConnection()->insertTablet(request); } catch (RedirectException& e) { } } else { @@ -1470,7 +1495,7 @@ void Session::insertTablet(Tablet& tablet, bool sorted) { void Session::insertRelationalTablet(Tablet& tablet, bool sorted) { std::unordered_map, Tablet> relationalTabletGroup; if (impl_->tableModelDeviceIdToEndpoint.empty()) { - relationalTabletGroup.insert(make_pair(impl_->defaultSessionConnection_, tablet)); + relationalTabletGroup.insert(make_pair(impl_->getDefaultSessionConnection(), tablet)); } else if (SessionUtils::isTabletContainsSingleDevice(tablet)) { relationalTabletGroup.insert( make_pair(impl_->getSessionConnection(tablet.getDeviceID(0)), tablet)); @@ -1571,7 +1596,7 @@ void Session::Impl::insertRelationalTabletOnce( removeBrokenSessionConnection(connection); try { TSStatus respStatus; - defaultSessionConnection_->getSessionClient()->insertTablet(respStatus, request); + getDefaultSessionConnection()->getSessionClient()->insertTablet(respStatus, request); RpcUtils::verifySuccess(respStatus); } catch (RedirectException& e) { } @@ -1693,7 +1718,7 @@ void Session::insertTablets(unordered_map& tablets, bool sorted request.__set_isAligned(isAligned); try { TSStatus respStatus; - impl_->defaultSessionConnection_->insertTablets(request); + impl_->getDefaultSessionConnection()->insertTablets(request); RpcUtils::verifySuccess(respStatus); } catch (RedirectException& e) { } @@ -1722,7 +1747,7 @@ void Session::testInsertRecord(const string& deviceId, int64_t time, req.__set_values(values); TSStatus tsStatus; try { - impl_->defaultSessionConnection_->testInsertStringRecord(req); + impl_->getDefaultSessionConnection()->testInsertStringRecord(req); RpcUtils::verifySuccess(tsStatus); } catch (const TTransportException& e) { log_debug(e.what()); @@ -1748,7 +1773,7 @@ void Session::testInsertTablet(const Tablet& tablet) { request.__set_size(tablet.rowSize); try { TSStatus tsStatus; - impl_->defaultSessionConnection_->testInsertTablet(request); + impl_->getDefaultSessionConnection()->testInsertTablet(request); RpcUtils::verifySuccess(tsStatus); } catch (const TTransportException& e) { log_debug(e.what()); @@ -1778,7 +1803,8 @@ void Session::testInsertRecords(const vector& deviceIds, const vectordefaultSessionConnection_->getSessionClient()->insertStringRecords(tsStatus, request); + impl_->getDefaultSessionConnection()->getSessionClient()->insertStringRecords(tsStatus, + request); RpcUtils::verifySuccess(tsStatus); } catch (const TTransportException& e) { log_debug(e.what()); @@ -1799,7 +1825,7 @@ void Session::deleteTimeseries(const string& path) { } void Session::deleteTimeseries(const vector& paths) { - impl_->defaultSessionConnection_->deleteTimeseries(paths); + impl_->getDefaultSessionConnection()->deleteTimeseries(paths); } void Session::deleteData(const string& path, int64_t endTime) { @@ -1817,11 +1843,11 @@ void Session::deleteData(const vector& paths, int64_t startTime, int64_t req.__set_paths(paths); req.__set_startTime(startTime); req.__set_endTime(endTime); - impl_->defaultSessionConnection_->deleteData(req); + impl_->getDefaultSessionConnection()->deleteData(req); } void Session::setStorageGroup(const string& storageGroupId) { - impl_->defaultSessionConnection_->setStorageGroup(storageGroupId); + impl_->getDefaultSessionConnection()->setStorageGroup(storageGroupId); } void Session::deleteStorageGroup(const string& storageGroup) { @@ -1831,7 +1857,7 @@ void Session::deleteStorageGroup(const string& storageGroup) { } void Session::deleteStorageGroups(const vector& storageGroups) { - impl_->defaultSessionConnection_->deleteStorageGroups(storageGroups); + impl_->getDefaultSessionConnection()->deleteStorageGroups(storageGroups); } void Session::createDatabase(const string& database) { @@ -1880,7 +1906,7 @@ void Session::createTimeseries(const string& path, TSDataType::TSDataType dataTy if (!measurementAlias.empty()) { req.__set_measurementAlias(measurementAlias); } - impl_->defaultSessionConnection_->createTimeseries(req); + impl_->getDefaultSessionConnection()->createTimeseries(req); } void Session::createMultiTimeseries(const vector& paths, @@ -1929,7 +1955,7 @@ void Session::createMultiTimeseries(const vector& paths, request.__set_measurementAliasList(*measurementAliasList); } - impl_->defaultSessionConnection_->createMultiTimeseries(request); + impl_->getDefaultSessionConnection()->createMultiTimeseries(request); } void Session::createAlignedTimeseries( @@ -1962,7 +1988,7 @@ void Session::createAlignedTimeseries( } request.__set_compressors(compressorsOrdinal); - impl_->defaultSessionConnection_->createAlignedTimeseries(request); + impl_->getDefaultSessionConnection()->createAlignedTimeseries(request); } bool Session::checkTimeseriesExists(const string& path) { @@ -1983,7 +2009,7 @@ bool Session::checkTimeseriesExists(const string& path) { shared_ptr Session::Impl::getQuerySessionConnection() { auto endPoint = nodesSupplier_->getQueryEndPoint(); if (!endPoint.is_initialized() || endPointToSessionConnection.empty()) { - return defaultSessionConnection_; + return getDefaultSessionConnection(); } auto it = endPointToSessionConnection.find(endPoint.value()); @@ -1991,6 +2017,7 @@ shared_ptr Session::Impl::getQuerySessionConnection() { return it->second; } + getDefaultSessionConnection(); shared_ptr newConnection; try { newConnection = @@ -2000,7 +2027,7 @@ shared_ptr Session::Impl::getQuerySessionConnection() { return newConnection; } catch (exception& e) { log_debug("Session::Impl::getQuerySessionConnection() exception: " + e.what()); - return newConnection; + return getDefaultSessionConnection(); } } @@ -2008,7 +2035,7 @@ shared_ptr Session::Impl::getSessionConnection(std::string de if (!enableRedirection_ || deviceIdToEndpoint.find(deviceId) == deviceIdToEndpoint.end() || endPointToSessionConnection.find(deviceIdToEndpoint[deviceId]) == endPointToSessionConnection.end()) { - return defaultSessionConnection_; + return getDefaultSessionConnection(); } return endPointToSessionConnection.find(deviceIdToEndpoint[deviceId])->second; } @@ -2019,21 +2046,21 @@ Session::Impl::getSessionConnection(std::shared_ptr deviceId tableModelDeviceIdToEndpoint.find(deviceId) == tableModelDeviceIdToEndpoint.end() || endPointToSessionConnection.find(tableModelDeviceIdToEndpoint[deviceId]) == endPointToSessionConnection.end()) { - return defaultSessionConnection_; + return getDefaultSessionConnection(); } return endPointToSessionConnection.find(tableModelDeviceIdToEndpoint[deviceId])->second; } string Session::getTimeZone() { - auto ret = impl_->defaultSessionConnection_->getTimeZone(); + auto ret = impl_->getDefaultSessionConnection()->getTimeZone(); return ret.timeZone; } void Session::setTimeZone(const string& zoneId) { TSSetTimeZoneReq req; - req.__set_sessionId(impl_->defaultSessionConnection_->sessionId); + req.__set_sessionId(impl_->getDefaultSessionConnection()->sessionId); req.__set_timeZone(zoneId); - impl_->defaultSessionConnection_->setTimeZone(req); + impl_->getDefaultSessionConnection()->setTimeZone(req); } unique_ptr Session::executeQueryStatement(const string& sql) { @@ -2047,6 +2074,7 @@ unique_ptr Session::executeQueryStatement(const string& sql, int void Session::Impl::handleQueryRedirection(TEndPoint endPoint) { if (!enableRedirection_) return; + getDefaultSessionConnection(); shared_ptr newConnection; auto it = endPointToSessionConnection.find(endPoint); if (it != endPointToSessionConnection.end()) { @@ -2070,6 +2098,7 @@ void Session::Impl::handleRedirection(const std::string& deviceId, TEndPoint end return; if (endPoint.ip == "0.0.0.0") return; + getDefaultSessionConnection(); deviceIdToEndpoint[deviceId] = endPoint; shared_ptr newConnection; @@ -2095,6 +2124,7 @@ void Session::Impl::handleRedirection(const std::shared_ptr& return; if (endPoint.ip == "0.0.0.0") return; + getDefaultSessionConnection(); tableModelDeviceIdToEndpoint[deviceId] = endPoint; shared_ptr newConnection; @@ -2127,7 +2157,7 @@ std::unique_ptr Session::executeQueryStatementMayRedirect(const log_warn("Session connection redirect exception: " + e.what()); impl_->handleQueryRedirection(endpointToThrift(e.endPoint)); try { - return impl_->defaultSessionConnection_->executeQueryStatement(sql, timeoutInMs); + return impl_->getDefaultSessionConnection()->executeQueryStatement(sql, timeoutInMs); } catch (exception& e) { log_error("Exception while executing redirected query statement: %s", e.what()); throw ExecutionException(e.what()); @@ -2140,7 +2170,9 @@ std::unique_ptr Session::executeQueryStatementMayRedirect(const void Session::executeNonQueryStatement(const string& sql) { try { - impl_->defaultSessionConnection_->executeNonQueryStatement(sql); + impl_->getDefaultSessionConnection()->executeNonQueryStatement(sql); + } catch (const IoTDBConnectionException&) { + throw; } catch (const exception& e) { throw IoTDBException(e.what()); } @@ -2148,7 +2180,7 @@ void Session::executeNonQueryStatement(const string& sql) { unique_ptr Session::executeRawDataQuery(const vector& paths, int64_t startTime, int64_t endTime) { - return impl_->defaultSessionConnection_->executeRawDataQuery(paths, startTime, endTime); + return impl_->getDefaultSessionConnection()->executeRawDataQuery(paths, startTime, endTime); } unique_ptr Session::executeLastDataQuery(const vector& paths) { @@ -2157,28 +2189,28 @@ unique_ptr Session::executeLastDataQuery(const vector& p unique_ptr Session::executeLastDataQuery(const vector& paths, int64_t lastTime) { - return impl_->defaultSessionConnection_->executeLastDataQuery(paths, lastTime); + return impl_->getDefaultSessionConnection()->executeLastDataQuery(paths, lastTime); } void Session::createSchemaTemplate(const Template& templ) { TSCreateSchemaTemplateReq req; req.__set_name(templ.getName()); req.__set_serializedTemplate(templ.serialize()); - impl_->defaultSessionConnection_->createSchemaTemplate(req); + impl_->getDefaultSessionConnection()->createSchemaTemplate(req); } void Session::setSchemaTemplate(const string& template_name, const string& prefix_path) { TSSetSchemaTemplateReq req; req.__set_templateName(template_name); req.__set_prefixPath(prefix_path); - impl_->defaultSessionConnection_->setSchemaTemplate(req); + impl_->getDefaultSessionConnection()->setSchemaTemplate(req); } void Session::unsetSchemaTemplate(const string& prefix_path, const string& template_name) { TSUnsetSchemaTemplateReq req; req.__set_templateName(template_name); req.__set_prefixPath(prefix_path); - impl_->defaultSessionConnection_->unsetSchemaTemplate(req); + impl_->getDefaultSessionConnection()->unsetSchemaTemplate(req); } void Session::addAlignedMeasurementsInTemplate( @@ -2212,7 +2244,7 @@ void Session::addAlignedMeasurementsInTemplate( } req.__set_compressors(compressorsOrdinal); - impl_->defaultSessionConnection_->appendSchemaTemplate(req); + impl_->getDefaultSessionConnection()->appendSchemaTemplate(req); } void Session::addAlignedMeasurementsInTemplate(const string& template_name, @@ -2258,7 +2290,7 @@ void Session::addUnalignedMeasurementsInTemplate( } req.__set_compressors(compressorsOrdinal); - impl_->defaultSessionConnection_->appendSchemaTemplate(req); + impl_->getDefaultSessionConnection()->appendSchemaTemplate(req); } void Session::addUnalignedMeasurementsInTemplate(const string& template_name, @@ -2278,14 +2310,14 @@ void Session::deleteNodeInTemplate(const string& template_name, const string& pa TSPruneSchemaTemplateReq req; req.__set_name(template_name); req.__set_path(path); - impl_->defaultSessionConnection_->pruneSchemaTemplate(req); + impl_->getDefaultSessionConnection()->pruneSchemaTemplate(req); } int Session::countMeasurementsInTemplate(const string& template_name) { TSQueryTemplateReq req; req.__set_name(template_name); req.__set_queryType(TemplateQueryType::COUNT_MEASUREMENTS); - TSQueryTemplateResp resp = impl_->defaultSessionConnection_->querySchemaTemplate(req); + TSQueryTemplateResp resp = impl_->getDefaultSessionConnection()->querySchemaTemplate(req); return resp.count; } @@ -2294,7 +2326,7 @@ bool Session::isMeasurementInTemplate(const string& template_name, const string& req.__set_name(template_name); req.__set_measurement(path); req.__set_queryType(TemplateQueryType::IS_MEASUREMENT); - TSQueryTemplateResp resp = impl_->defaultSessionConnection_->querySchemaTemplate(req); + TSQueryTemplateResp resp = impl_->getDefaultSessionConnection()->querySchemaTemplate(req); return resp.result; } @@ -2303,7 +2335,7 @@ bool Session::isPathExistInTemplate(const string& template_name, const string& p req.__set_name(template_name); req.__set_measurement(path); req.__set_queryType(TemplateQueryType::PATH_EXIST); - TSQueryTemplateResp resp = impl_->defaultSessionConnection_->querySchemaTemplate(req); + TSQueryTemplateResp resp = impl_->getDefaultSessionConnection()->querySchemaTemplate(req); return resp.result; } @@ -2312,7 +2344,7 @@ std::vector Session::showMeasurementsInTemplate(const string& templ req.__set_name(template_name); req.__set_measurement(""); req.__set_queryType(TemplateQueryType::SHOW_MEASUREMENTS); - TSQueryTemplateResp resp = impl_->defaultSessionConnection_->querySchemaTemplate(req); + TSQueryTemplateResp resp = impl_->getDefaultSessionConnection()->querySchemaTemplate(req); return resp.measurements; } @@ -2322,7 +2354,7 @@ std::vector Session::showMeasurementsInTemplate(const string& templ req.__set_name(template_name); req.__set_measurement(pattern); req.__set_queryType(TemplateQueryType::SHOW_MEASUREMENTS); - TSQueryTemplateResp resp = impl_->defaultSessionConnection_->querySchemaTemplate(req); + TSQueryTemplateResp resp = impl_->getDefaultSessionConnection()->querySchemaTemplate(req); return resp.measurements; } diff --git a/iotdb-client/client-cpp/src/session/SessionC.cpp b/iotdb-client/client-cpp/src/session/SessionC.cpp index 7d86bd348f851..79287cf025235 100644 --- a/iotdb-client/client-cpp/src/session/SessionC.cpp +++ b/iotdb-client/client-cpp/src/session/SessionC.cpp @@ -26,11 +26,12 @@ #include "SessionDataSet.h" #include -#include -#include #include -#include #include +#include +#include +#include +#include /* ============================================================ * Internal wrapper structs — the opaque handles point to these @@ -87,6 +88,10 @@ static TsStatus handleException(const std::exception& e) { dynamic_cast(&e)) { return setError(TS_ERR_EXECUTION, e); } + if (dynamic_cast(&e) || + dynamic_cast(&e)) { + return setError(TS_ERR_INVALID_PARAM, e); + } #endif return setError(TS_ERR_UNKNOWN, e); } @@ -678,7 +683,10 @@ TsStatus ts_tablet_set_row_count(CTablet* tablet, int rowCount) { clearError(); if (!tablet) return setError(TS_ERR_NULL_PTR, "tablet is null"); - tablet->cpp.rowSize = rowCount; + if (rowCount < 0 || static_cast(rowCount) > tablet->cpp.maxRowNumber) { + return setError(TS_ERR_INVALID_PARAM, "rowCount out of range [0, maxRowNumber]"); + } + tablet->cpp.rowSize = static_cast(rowCount); return TS_OK; } diff --git a/iotdb-client/client-cpp/test/cpp/sessionCIT.cpp b/iotdb-client/client-cpp/test/cpp/sessionCIT.cpp index 9db5361635d54..0e0791c5d2943 100644 --- a/iotdb-client/client-cpp/test/cpp/sessionCIT.cpp +++ b/iotdb-client/client-cpp/test/cpp/sessionCIT.cpp @@ -497,6 +497,27 @@ TEST_CASE("C API - Tablet row count and reset", "[c_tabletReset]") { ts_tablet_destroy(tablet); } +TEST_CASE("C API - Tablet index bounds", "[c_tabletBounds]") { + CaseReporter cr("c_tabletBounds"); + const char* colNames[] = {"s1"}; + TSDataType_C dts[] = {TS_TYPE_INT64}; + CTablet* tablet = ts_tablet_new("root.ctest.d1", 1, colNames, dts, 10); + REQUIRE(tablet != nullptr); + + REQUIRE(ts_tablet_add_timestamp(tablet, 0, 1) == TS_OK); + REQUIRE(ts_tablet_add_value_int64(tablet, 0, 0, 100) == TS_OK); + REQUIRE(ts_tablet_add_timestamp(tablet, 100, 2) != TS_OK); + REQUIRE(ts_tablet_add_value_int64(tablet, 1, 0, 100) != TS_OK); + REQUIRE(ts_tablet_add_value_int64(tablet, 0, 100, 100) != TS_OK); + REQUIRE(ts_tablet_set_row_count(tablet, -1) != TS_OK); + REQUIRE(ts_tablet_set_row_count(tablet, 11) != TS_OK); + REQUIRE(ts_tablet_set_row_count(tablet, 1000000) != TS_OK); + REQUIRE(ts_tablet_set_row_count(tablet, 5) == TS_OK); + REQUIRE(ts_tablet_get_row_count(tablet) == 5); + + ts_tablet_destroy(tablet); +} + TEST_CASE("C API - Aligned timeseries and aligned writes", "[c_aligned]") { CaseReporter cr("c_aligned"); diff --git a/iotdb-client/client-cpp/test/cpp/sessionCRelationalIT.cpp b/iotdb-client/client-cpp/test/cpp/sessionCRelationalIT.cpp index 9a78f2512ed44..4a298dd1c1a13 100644 --- a/iotdb-client/client-cpp/test/cpp/sessionCRelationalIT.cpp +++ b/iotdb-client/client-cpp/test/cpp/sessionCRelationalIT.cpp @@ -254,6 +254,19 @@ TEST_CASE("C API Table - Multi-node table session", "[c_table_multiNode][c_table ts_table_session_destroy(localSession); } +TEST_CASE("C API Table - Reject SQL after close", "[c_table_close][c_table_lifecycle]") { + CaseReporter cr("c_table_close"); + + CTableSession* localSession = ts_table_session_new("127.0.0.1", 6667, "root", "root", ""); + REQUIRE(localSession != nullptr); + REQUIRE(ts_table_session_open(localSession) == TS_OK); + + ts_table_session_close(localSession); + + REQUIRE(ts_table_session_execute_non_query(localSession, "SHOW DATABASES") != TS_OK); + ts_table_session_destroy(localSession); +} + /* ============================================================ * Dataset column info (table model) * ============================================================ */ diff --git a/iotdb-client/client-cpp/test/cpp/sessionIT.cpp b/iotdb-client/client-cpp/test/cpp/sessionIT.cpp index c0428f6e84a75..3b19f2e2b25d6 100644 --- a/iotdb-client/client-cpp/test/cpp/sessionIT.cpp +++ b/iotdb-client/client-cpp/test/cpp/sessionIT.cpp @@ -28,6 +28,7 @@ #include "common_types.h" #include +#include #include using namespace std; @@ -369,6 +370,30 @@ TEST_CASE("Test insertRecordsOfOneDevice", "[testInsertRecordsOfOneDevice]") { REQUIRE(count == 100); } +TEST_CASE("Tablet index bounds", "[tabletBounds]") { + CaseReporter cr("tabletBounds"); + vector> schemaList; + schemaList.emplace_back("s1", TSDataType::INT64); + Tablet tablet("root.test.d1", schemaList, 10); + tablet.addTimestamp(0, 1); + tablet.addValue(0, 0, static_cast(100)); + + REQUIRE_THROWS_AS(tablet.addTimestamp(10, 11), out_of_range); + REQUIRE_THROWS_AS(tablet.addValue(1, 0, static_cast(1)), out_of_range); + REQUIRE_THROWS_AS(tablet.addValue(0, 100, static_cast(1)), out_of_range); +} + +TEST_CASE("Session rejects SQL after close", "[sessionClose]") { + CaseReporter cr("sessionClose"); + SessionBuilder builder; + auto localSession = + builder.host("127.0.0.1")->rpcPort(6667)->username("root")->password("root")->build(); + localSession->open(); + localSession->close(); + REQUIRE_THROWS_AS(localSession->executeNonQueryStatement("show databases"), + IoTDBConnectionException); +} + TEST_CASE("Test insertTablet ", "[testInsertTablet]") { CaseReporter cr("testInsertTablet"); prepareTimeseries(); diff --git a/iotdb-client/client-cpp/test/cpp/sessionRelationalIT.cpp b/iotdb-client/client-cpp/test/cpp/sessionRelationalIT.cpp index 461a3ec863055..9ed3d334b1c4a 100644 --- a/iotdb-client/client-cpp/test/cpp/sessionRelationalIT.cpp +++ b/iotdb-client/client-cpp/test/cpp/sessionRelationalIT.cpp @@ -86,6 +86,19 @@ TEST_CASE("Test TableSession builder with nodeUrls", "[SessionBuilderInit]") { session->close(); } +TEST_CASE("TableSession rejects SQL after close", "[tableSessionClose]") { + CaseReporter cr("tableSessionClose"); + + TableSessionBuilder builder; + auto localSession = + builder.host("127.0.0.1")->rpcPort(6667)->username("root")->password("root")->build(); + localSession->open(); + localSession->close(); + + REQUIRE_THROWS_AS(localSession->executeNonQueryStatement("SHOW DATABASES"), + IoTDBConnectionException); +} + TEST_CASE("Test insertRelationalTablet", "[testInsertRelationalTablet]") { CaseReporter cr("testInsertRelationalTablet"); session->executeNonQueryStatement("CREATE DATABASE IF NOT EXISTS db1"); 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/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..d2dee00f29aca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java @@ -32,6 +32,7 @@ import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.process.SingleChildProcessNode; import org.apache.iotdb.commons.queryengine.plan.relational.function.BoundSignature; import org.apache.iotdb.commons.queryengine.plan.relational.function.FunctionId; +import org.apache.iotdb.commons.queryengine.plan.relational.function.TableBuiltinTableFunction; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ResolvedFunction; @@ -1826,22 +1827,34 @@ public List visitTableFunctionProcessor( if (node.getChildren().isEmpty()) { return Collections.singletonList(node); } - boolean canSplitPushDown = node.isRowSemantic() || (node.getChild() instanceof GroupNode); + boolean canSplitPushDown = canSplitTableFunctionProcessor(node); List childrenNodes = node.getChild().accept(this, context); if (childrenNodes.size() == 1) { node.setChild(childrenNodes.get(0)); return Collections.singletonList(node); } else if (!canSplitPushDown) { - CollectNode collectNode = - new CollectNode(queryId.genPlanNodeId(), node.getChildren().get(0).getOutputSymbols()); - childrenNodes.forEach(collectNode::addChild); - node.setChild(collectNode); + OrderingScheme childOrdering = nodeOrderingMap.get(childrenNodes.get(0).getPlanNodeId()); + node.setChild(mergeChildrenViaCollectOrMergeSort(childOrdering, childrenNodes)); return Collections.singletonList(node); } else { return splitForEachChild(node, childrenNodes); } } + private boolean canSplitTableFunctionProcessor(TableFunctionProcessorNode node) { + if (node.isRowSemantic()) { + return true; + } + if (!isPartitionedGroup(node.getChild())) { + return false; + } + return !TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(node.getName()); + } + + private boolean isPartitionedGroup(PlanNode node) { + return node instanceof GroupNode && ((GroupNode) node).getPartitionKeyCount() > 0; + } + private void buildRegionNodeMap( AggregationTableScanNode originalAggTableScanNode, List> 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 9d09ff057b30a..637113026f4f0 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,179 @@ 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", "czE=,czI=,czM=") + .build()); + + assertPlan( + logicalQueryPlan, anyTree(tableFunctionProcessor(tableFunctionMatcher, sort(tableScan)))); + assertPlan( + planTester.getFragmentPlan(0), + output( + tableFunctionProcessor( + tableFunctionMatcher, mergeSort(exchange(), exchange(), exchange())))); + assertPlan(planTester.getFragmentPlan(1), sort(tableScan)); + assertPlan(planTester.getFragmentPlan(2), sort(tableScan)); + assertPlan(planTester.getFragmentPlan(3), sort(tableScan)); + } + + @Test + public void testFFTDefaultArguments() { + PlanTester planTester = new PlanTester(); + String sql = "SELECT * FROM TABLE(FFT(DATA => TABLE(table1) ORDER BY time))"; + LogicalQueryPlan logicalQueryPlan = planTester.createPlan(sql); + PlanMatchPattern tableScan = + tableScan( + "testdb.table1", + ImmutableMap.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", "czE=,czI=,czM=") + .build()); + + assertPlan( + logicalQueryPlan, anyTree(tableFunctionProcessor(tableFunctionMatcher, sort(tableScan)))); + assertPlan( + planTester.getFragmentPlan(0), + output( + tableFunctionProcessor( + tableFunctionMatcher, mergeSort(exchange(), exchange(), exchange())))); + assertPlan(planTester.getFragmentPlan(1), sort(tableScan)); + assertPlan(planTester.getFragmentPlan(2), sort(tableScan)); + assertPlan(planTester.getFragmentPlan(3), sort(tableScan)); + } + + @Test + public void testFFTWithSpecifiedTimeColumn() { + PlanTester planTester = new PlanTester(); + String sql = + "SELECT * FROM FFT(" + + "DATA => (SELECT time AS event_time, tag1, s1 FROM table1) " + + "PARTITION BY tag1 ORDER BY event_time, " + + "TIMECOL => 'event_time', " + + "SAMPLE_INTERVAL => 1s, " + + "N => 4)"; + + planTester.createPlan(sql); + } + + @Test + public void testFFTPositionalArgumentsKeepExistingOrder() { + PlanTester planTester = new PlanTester(); + String sql = "SELECT * FROM TABLE(FFT(TABLE(table1) ORDER BY time, 1s, 4, 'ortho'))"; + + planTester.createPlan(sql); + } + + @Test + public void testFFTRejectsInvalidArguments() { + assertAnalyzeFails( + "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1, SAMPLE_INTERVAL => 1ms)", + "Table argument with set semantics requires an ORDER BY clause."); + assertAnalyzeFails( + "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time DESC, SAMPLE_INTERVAL => 1ms)", + "The ORDER BY clause of the DATA argument must sort the time column in ascending order."); + assertAnalyzeFails( + "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY s1, SAMPLE_INTERVAL => 1ms)", + "The ORDER BY clause of the DATA argument must contain exactly the time column specified by the TIMECOL argument."); + assertAnalyzeFails( + "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time, SAMPLE_INTERVAL => 1)", + "The SAMPLE_INTERVAL argument of FFT must be a duration literal."); + assertAnalyzeFails( + "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time, N => 0)", + "Invalid scalar argument N, should be a positive value"); + assertAnalyzeFails( + "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time, N => 65537)", + "FFT transform length N must not exceed 65536."); + assertAnalyzeFails( + "SELECT * FROM FFT(DATA => table1 PARTITION BY tag1 ORDER BY time, NORM => 'bad')", + "Invalid NORM value for FFT. Supported values are backward, forward and ortho."); + assertAnalyzeFails( + "SELECT * FROM FFT(DATA => (SELECT time, tag1 FROM table1) PARTITION BY tag1 ORDER BY time)", + "No numeric columns found for FFT calculation."); + } + + private void assertAnalyzeFails(String sql, String message) { + try { + analyzeSQL(sql, TEST_MATADATA, QUERY_CONTEXT); + fail(); + } catch (SemanticException e) { + assertEquals(message, e.getMessage()); + } + } + @Test public void testM4TimeWindowMode() { PlanTester planTester = new PlanTester(); diff --git a/iotdb-core/node-commons/src/main/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..f88f6b48cbf7b --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunction.java @@ -0,0 +1,804 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.udf.builtin.relational.tvf; + +import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils; +import org.apache.iotdb.commons.udf.builtin.relational.tvf.fft.DoubleFFT_1D; +import org.apache.iotdb.commons.udf.builtin.relational.tvf.fft.FloatFFT_1D; +import org.apache.iotdb.udf.api.exception.UDFException; +import org.apache.iotdb.udf.api.relational.TableFunction; +import org.apache.iotdb.udf.api.relational.access.Record; +import org.apache.iotdb.udf.api.relational.table.MapTableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis; +import org.apache.iotdb.udf.api.relational.table.TableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.TableFunctionProcessorProvider; +import org.apache.iotdb.udf.api.relational.table.argument.Argument; +import org.apache.iotdb.udf.api.relational.table.argument.DescribedSchema; +import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument; +import org.apache.iotdb.udf.api.relational.table.argument.TableArgument; +import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor; +import org.apache.iotdb.udf.api.relational.table.specification.ParameterSpecification; +import org.apache.iotdb.udf.api.relational.table.specification.ScalarParameterSpecification; +import org.apache.iotdb.udf.api.relational.table.specification.TableParameterSpecification; +import org.apache.iotdb.udf.api.type.Type; + +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.utils.Binary; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Base64; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +import static org.apache.iotdb.commons.udf.builtin.relational.tvf.WindowTVFUtils.findColumnIndex; +import static org.apache.iotdb.udf.api.relational.table.argument.ScalarArgumentChecker.POSITIVE_LONG_CHECKER; + +public class FFTTableFunction implements TableFunction { + + public static final String DATA_PARAMETER_NAME = "DATA"; + public static final String TIMECOL_PARAMETER_NAME = "TIMECOL"; + public static final String SAMPLE_INTERVAL_PARAMETER_NAME = "SAMPLE_INTERVAL"; + public static final String N_PARAMETER_NAME = "N"; + public static final String NORM_PARAMETER_NAME = "NORM"; + public static final String SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME = + "__FFT_SAMPLE_INTERVAL_SPECIFIED"; + + private static final String DEFAULT_TIME_COLUMN_NAME = "time"; + private static final String OUTPUT_FREQUENCY_INDEX_COLUMN = "frequency_index"; + private static final String OUTPUT_FREQUENCY_COLUMN = "frequency"; + private static final String PARTITION_TYPES_PROPERTY = "__FFT_PARTITION_TYPES"; + private static final String VALUE_TYPES_PROPERTY = "__FFT_VALUE_TYPES"; + private static final String VALUE_NAMES_PROPERTY = "__FFT_VALUE_NAMES"; + private static final long UNSPECIFIED_SAMPLE_INTERVAL = Long.MIN_VALUE; + private static final long UNSPECIFIED_N = -1L; + private static final long MAX_TRANSFORM_LENGTH = 65_536L; + private static final long MAX_SPECTRUM_VALUES = 16_777_216L; + private static final String NORM_BACKWARD = "backward"; + private static final String NORM_FORWARD = "forward"; + private static final String NORM_ORTHO = "ortho"; + private static final Set 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(), + ScalarParameterSpecification.builder() + .name(TIMECOL_PARAMETER_NAME) + .type(Type.STRING) + .defaultValue(DEFAULT_TIME_COLUMN_NAME) + .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."); + } + + String timeColumn = + (String) ((ScalarArgument) arguments.get(TIMECOL_PARAMETER_NAME)).getValue(); + int timeColumnIndex = + findColumnIndex(tableArgument, timeColumn, Collections.singleton(Type.TIMESTAMP)); + validateOrderBy(tableArgument, timeColumn); + + 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, encodeStrings(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 = decodeStrings((String) handle.getProperty(VALUE_NAMES_PROPERTY)); + + return new TableFunctionProcessorProvider() { + @Override + public TableFunctionDataProcessor getDataProcessor() { + return new FFTDataProcessor( + sampleIntervalSpecified, + sampleInterval, + transformLength, + norm, + createColumns(partitionTypes, 1), + createNumericColumns(valueTypes, valueNames, partitionTypes.length + 1)); + } + }; + } + + private static void validateOrderBy(TableArgument tableArgument, String timeColumn) { + if (tableArgument.getOrderBy().size() != 1 + || !tableArgument.getOrderBy().get(0).equalsIgnoreCase(timeColumn)) { + throw new SemanticException( + "The ORDER BY clause of the DATA argument must contain exactly the time column specified by the TIMECOL argument."); + } + } + + 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 spectrumValues; + try { + 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 (spectrumValues > MAX_SPECTRUM_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 encodeStrings(List values) { + StringBuilder builder = new StringBuilder(); + for (int i = 0; i < values.size(); i++) { + if (i > 0) { + builder.append(','); + } + builder.append( + Base64.getEncoder().encodeToString(values.get(i).getBytes(StandardCharsets.UTF_8))); + } + return builder.toString(); + } + + private static String[] decodeStrings(String value) { + if (value.isEmpty()) { + return new String[0]; + } + String[] encodedValues = value.split(","); + String[] decodedValues = new String[encodedValues.length]; + for (int i = 0; i < encodedValues.length; i++) { + decodedValues[i] = + new String(Base64.getDecoder().decode(encodedValues[i]), StandardCharsets.UTF_8); + } + return decodedValues; + } + + private static ValueColumn[] createColumns(Type[] types, int firstInputIndex) { + ValueColumn[] columns = new ValueColumn[types.length]; + for (int i = 0; i < types.length; i++) { + columns[i] = new ValueColumn(firstInputIndex + i, ValueOperator.fromType(types[i])); + } + return columns; + } + + private static NumericColumn[] createNumericColumns( + Type[] types, String[] names, int firstInputIndex) { + NumericColumn[] columns = new NumericColumn[types.length]; + for (int i = 0; i < types.length; i++) { + columns[i] = + new NumericColumn(firstInputIndex + i, names[i], NumericOperator.fromType(types[i])); + } + return columns; + } + + private enum ValueOperator { + BOOLEAN(Type.BOOLEAN) { + @Override + Object read(Record record, int index) { + return record.getBoolean(index); + } + + @Override + void write(ColumnBuilder builder, Object value) { + builder.writeBoolean((Boolean) value); + } + }, + INT32(Type.INT32) { + @Override + Object read(Record record, int index) { + return record.getInt(index); + } + + @Override + void write(ColumnBuilder builder, Object value) { + builder.writeInt((Integer) value); + } + }, + INT64(Type.INT64) { + @Override + Object read(Record record, int index) { + return record.getLong(index); + } + + @Override + void write(ColumnBuilder builder, Object value) { + builder.writeLong((Long) value); + } + }, + FLOAT(Type.FLOAT) { + @Override + Object read(Record record, int index) { + return record.getFloat(index); + } + + @Override + void write(ColumnBuilder builder, Object value) { + builder.writeFloat((Float) value); + } + }, + DOUBLE(Type.DOUBLE) { + @Override + Object read(Record record, int index) { + return record.getDouble(index); + } + + @Override + void write(ColumnBuilder builder, Object value) { + builder.writeDouble((Double) value); + } + }, + TEXT(Type.TEXT) { + @Override + Object read(Record record, int index) { + return record.getBinary(index); + } + + @Override + void write(ColumnBuilder builder, Object value) { + builder.writeBinary((Binary) value); + } + }, + BLOB(Type.BLOB) { + @Override + Object read(Record record, int index) { + return record.getBinary(index); + } + + @Override + void write(ColumnBuilder builder, Object value) { + builder.writeBinary((Binary) value); + } + }, + TIMESTAMP(Type.TIMESTAMP) { + @Override + Object read(Record record, int index) { + return record.getLong(index); + } + + @Override + void write(ColumnBuilder builder, Object value) { + builder.writeLong((Long) value); + } + }, + DATE(Type.DATE) { + @Override + Object read(Record record, int index) { + return record.getLocalDate(index); + } + + @Override + void write(ColumnBuilder builder, Object value) { + builder.writeObject(value); + } + }, + STRING(Type.STRING) { + @Override + Object read(Record record, int index) { + return record.getBinary(index); + } + + @Override + void write(ColumnBuilder builder, Object value) { + builder.writeBinary((Binary) value); + } + }; + + private final Type type; + + ValueOperator(Type type) { + this.type = type; + } + + abstract Object read(Record record, int index); + + abstract void write(ColumnBuilder builder, Object value); + + static ValueOperator fromType(Type type) { + for (ValueOperator valueOperator : values()) { + if (valueOperator.type == type) { + return valueOperator; + } + } + throw new IllegalArgumentException("Unsupported FFT partition type: " + type); + } + } + + private enum NumericOperator { + INT32(Type.INT32, false) { + @Override + Number read(Record record, int index) { + return record.getInt(index); + } + }, + INT64(Type.INT64, false) { + @Override + Number read(Record record, int index) { + return record.getLong(index); + } + }, + FLOAT(Type.FLOAT, true) { + @Override + Number read(Record record, int index) { + return record.getFloat(index); + } + }, + DOUBLE(Type.DOUBLE, false) { + @Override + Number read(Record record, int index) { + return record.getDouble(index); + } + }; + + private final Type type; + private final boolean floatFft; + + NumericOperator(Type type, boolean floatFft) { + this.type = type; + this.floatFft = floatFft; + } + + abstract Number read(Record record, int index); + + boolean usesFloatFft() { + return floatFft; + } + + static NumericOperator fromType(Type type) { + for (NumericOperator numericOperator : values()) { + if (numericOperator.type == type) { + return numericOperator; + } + } + throw new IllegalArgumentException("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 Number read(Record record) { + return numericOperator.read(record, inputIndex); + } + + private boolean usesFloatFft() { + return numericOperator.usesFloatFft(); + } + } + + private static class Spectrum { + private final double[] doubleValues; + private final float[] floatValues; + + private Spectrum(double[] doubleValues, float[] floatValues) { + this.doubleValues = doubleValues; + this.floatValues = floatValues; + } + + private static Spectrum fromDouble(double[] values) { + return new Spectrum(values, null); + } + + private static Spectrum fromFloat(float[] values) { + return new Spectrum(null, values); + } + + private double real(int frequencyIndex, double scaleFactor) { + int index = 2 * frequencyIndex; + return (doubleValues == null ? floatValues[index] : doubleValues[index]) * scaleFactor; + } + + private double imaginary(int frequencyIndex, double scaleFactor) { + int index = 2 * frequencyIndex + 1; + return (doubleValues == null ? floatValues[index] : doubleValues[index]) * scaleFactor; + } + } + + private static class FFTDataProcessor implements TableFunctionDataProcessor { + private final boolean sampleIntervalSpecified; + private final long sampleInterval; + private final long specifiedTransformLength; + private final String norm; + private final ValueColumn[] partitionColumns; + private final NumericColumn[] valueColumns; + private final Object[] partitionValues; + private final boolean[] partitionValueIsNull; + private final List rows = new ArrayList<>(); + private long inputRowCount; + private long firstTime; + private long previousTime; + private boolean initialized; + + private FFTDataProcessor( + boolean sampleIntervalSpecified, + long sampleInterval, + long specifiedTransformLength, + String norm, + ValueColumn[] partitionColumns, + NumericColumn[] valueColumns) { + this.sampleIntervalSpecified = sampleIntervalSpecified; + this.sampleInterval = sampleInterval; + this.specifiedTransformLength = specifiedTransformLength; + this.norm = norm; + this.partitionColumns = partitionColumns; + this.valueColumns = valueColumns; + this.partitionValues = new Object[partitionColumns.length]; + this.partitionValueIsNull = new boolean[partitionColumns.length]; + } + + @Override + public void process( + Record input, + List properColumnBuilders, + ColumnBuilder passThroughIndexBuilder) { + 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."); + } + previousTime = currentTime; + + if (specifiedTransformLength == UNSPECIFIED_N) { + validateTransformLength(inputRowCount + 1, valueColumns.length); + } + + boolean shouldCacheRow = + specifiedTransformLength == UNSPECIFIED_N || rows.size() < specifiedTransformLength; + Number[] row = shouldCacheRow ? new Number[valueColumns.length] : null; + for (int i = 0; i < valueColumns.length; i++) { + NumericColumn valueColumn = valueColumns[i]; + if (input.isNull(valueColumn.inputIndex)) { + throw new SemanticException( + String.format("FFT does not support null values in column [%s].", valueColumn.name)); + } + if (shouldCacheRow) { + row[i] = valueColumn.read(input); + } + } + inputRowCount++; + if (shouldCacheRow) { + rows.add(row); + } + } + + @Override + public void finish( + List properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { + if (inputRowCount == 0) { + return; + } + + int transformLength = getTransformLength(); + double sampleIntervalSeconds = getSampleIntervalSeconds(); + double scaleFactor = getScaleFactor(transformLength); + Spectrum[] spectra = new Spectrum[valueColumns.length]; + + int copiedRows = Math.min(rows.size(), transformLength); + DoubleFFT_1D doubleFft = null; + FloatFFT_1D floatFft = null; + for (int columnIndex = 0; columnIndex < valueColumns.length; columnIndex++) { + if (valueColumns[columnIndex].usesFloatFft()) { + float[] spectrum = new float[2 * transformLength]; + for (int rowIndex = 0; rowIndex < copiedRows; rowIndex++) { + spectrum[2 * rowIndex] = rows.get(rowIndex)[columnIndex].floatValue(); + } + if (floatFft == null) { + floatFft = new FloatFFT_1D(transformLength); + } + floatFft.complexForward(spectrum); + spectra[columnIndex] = Spectrum.fromFloat(spectrum); + } else { + double[] spectrum = new double[2 * transformLength]; + for (int rowIndex = 0; rowIndex < copiedRows; rowIndex++) { + spectrum[2 * rowIndex] = rows.get(rowIndex)[columnIndex].doubleValue(); + } + if (doubleFft == null) { + doubleFft = new DoubleFFT_1D(transformLength); + } + doubleFft.complexForward(spectrum); + spectra[columnIndex] = Spectrum.fromDouble(spectrum); + } + } + + for (int frequencyIndex = 0; frequencyIndex < transformLength; frequencyIndex++) { + int outputColumnIndex = 0; + for (int partitionIndex = 0; partitionIndex < partitionColumns.length; partitionIndex++) { + if (partitionValueIsNull[partitionIndex]) { + properColumnBuilders.get(outputColumnIndex++).appendNull(); + } else { + partitionColumns[partitionIndex].write( + properColumnBuilders.get(outputColumnIndex++), partitionValues[partitionIndex]); + } + } + properColumnBuilders.get(outputColumnIndex++).writeLong(frequencyIndex); + properColumnBuilders + .get(outputColumnIndex++) + .writeDouble( + calculateFrequency(frequencyIndex, transformLength, sampleIntervalSeconds)); + for (int columnIndex = 0; columnIndex < valueColumns.length; columnIndex++) { + properColumnBuilders + .get(outputColumnIndex++) + .writeDouble(spectra[columnIndex].real(frequencyIndex, scaleFactor)); + properColumnBuilders + .get(outputColumnIndex++) + .writeDouble(spectra[columnIndex].imaginary(frequencyIndex, scaleFactor)); + } + } + } + + private void capturePartitionValues(Record input) { + for (int i = 0; i < partitionColumns.length; i++) { + if (input.isNull(partitionColumns[i].inputIndex)) { + partitionValueIsNull[i] = true; + } else { + partitionValues[i] = partitionColumns[i].read(input); + } + } + } + + private int getTransformLength() { + long transformLength = + specifiedTransformLength == UNSPECIFIED_N ? inputRowCount : specifiedTransformLength; + validateTransformLength(transformLength, valueColumns.length); + return (int) transformLength; + } + + private double getSampleIntervalSeconds() { + double interval; + if (sampleIntervalSpecified) { + interval = sampleInterval; + } else { + if (inputRowCount < 2) { + throw new SemanticException("FFT requires at least two rows to infer SAMPLE_INTERVAL."); + } + interval = (double) (previousTime - firstTime) / (inputRowCount - 1); + } + 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 double getScaleFactor(int transformLength) { + if (NORM_FORWARD.equals(norm)) { + return 1.0 / transformLength; + } + if (NORM_ORTHO.equals(norm)) { + return 1.0 / Math.sqrt(transformLength); + } + return 1.0; + } + + private double calculateFrequency( + int frequencyIndex, int transformLength, double sampleIntervalSeconds) { + int positiveFrequencyCount = (transformLength + 1) / 2; + int signedIndex = + frequencyIndex < positiveFrequencyCount + ? frequencyIndex + : frequencyIndex - transformLength; + return signedIndex / (transformLength * sampleIntervalSeconds); + } + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/DoubleFFT_1D.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/DoubleFFT_1D.java new file mode 100644 index 0000000000000..f1a2a6d71a48d --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/DoubleFFT_1D.java @@ -0,0 +1,155 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.udf.builtin.relational.tvf.fft; + +/** Computes an in-place 1D forward DFT for interleaved complex double data. */ +public final class DoubleFFT_1D { + + private final int length; + + public DoubleFFT_1D(long length) { + if (length < 1 || length > Integer.MAX_VALUE) { + throw new IllegalArgumentException("FFT length must be a positive int-sized value."); + } + this.length = (int) length; + } + + public void complexForward(double[] values) { + if (values.length < 2 * length) { + throw new IllegalArgumentException("Input array length must be at least 2 * FFT length."); + } + if (length == 1) { + return; + } + if (isPowerOfTwo(length)) { + transform(values, length, false); + } else { + bluesteinForward(values); + } + } + + private void bluesteinForward(double[] values) { + int convolutionLength = nextPowerOfTwo(2 * length - 1); + double[] a = new double[2 * convolutionLength]; + double[] b = new double[2 * convolutionLength]; + + for (int i = 0; i < length; i++) { + double angle = Math.PI * i * (double) i / length; + double cos = Math.cos(angle); + double sin = Math.sin(angle); + double real = values[2 * i]; + double imaginary = values[2 * i + 1]; + + a[2 * i] = real * cos + imaginary * sin; + a[2 * i + 1] = imaginary * cos - real * sin; + b[2 * i] = cos; + b[2 * i + 1] = sin; + if (i > 0) { + b[2 * (convolutionLength - i)] = cos; + b[2 * (convolutionLength - i) + 1] = sin; + } + } + + transform(a, convolutionLength, false); + transform(b, convolutionLength, false); + for (int i = 0; i < convolutionLength; i++) { + int offset = 2 * i; + double real = a[offset] * b[offset] - a[offset + 1] * b[offset + 1]; + double imaginary = a[offset] * b[offset + 1] + a[offset + 1] * b[offset]; + a[offset] = real; + a[offset + 1] = imaginary; + } + transform(a, convolutionLength, true); + + for (int i = 0; i < length; i++) { + double angle = Math.PI * i * (double) i / length; + double cos = Math.cos(angle); + double sin = Math.sin(angle); + double real = a[2 * i]; + double imaginary = a[2 * i + 1]; + values[2 * i] = real * cos + imaginary * sin; + values[2 * i + 1] = imaginary * cos - real * sin; + } + } + + private static void transform(double[] values, int size, boolean inverse) { + for (int i = 1, j = 0; i < size; i++) { + int bit = size >>> 1; + while ((j & bit) != 0) { + j ^= bit; + bit >>>= 1; + } + j ^= bit; + if (i < j) { + swap(values, 2 * i, 2 * j); + swap(values, 2 * i + 1, 2 * j + 1); + } + } + + for (int step = 2; step <= size; step <<= 1) { + double angle = (inverse ? 2.0 : -2.0) * Math.PI / step; + double stepReal = Math.cos(angle); + double stepImaginary = Math.sin(angle); + int halfStep = step >>> 1; + for (int block = 0; block < size; block += step) { + double factorReal = 1.0; + double factorImaginary = 0.0; + for (int j = 0; j < halfStep; j++) { + int even = 2 * (block + j); + int odd = 2 * (block + j + halfStep); + double oddReal = values[odd] * factorReal - values[odd + 1] * factorImaginary; + double oddImaginary = values[odd] * factorImaginary + values[odd + 1] * factorReal; + double evenReal = values[even]; + double evenImaginary = values[even + 1]; + + values[even] = evenReal + oddReal; + values[even + 1] = evenImaginary + oddImaginary; + values[odd] = evenReal - oddReal; + values[odd + 1] = evenImaginary - oddImaginary; + + double nextFactorReal = factorReal * stepReal - factorImaginary * stepImaginary; + factorImaginary = factorReal * stepImaginary + factorImaginary * stepReal; + factorReal = nextFactorReal; + } + } + } + + if (inverse) { + for (int i = 0; i < 2 * size; i++) { + values[i] /= size; + } + } + } + + private static boolean isPowerOfTwo(int value) { + return (value & (value - 1)) == 0; + } + + private static int nextPowerOfTwo(int value) { + int highestOneBit = Integer.highestOneBit(value); + return value == highestOneBit ? value : highestOneBit << 1; + } + + private static void swap(double[] values, int left, int right) { + double tmp = values[left]; + values[left] = values[right]; + values[right] = tmp; + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FloatFFT_1D.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FloatFFT_1D.java new file mode 100644 index 0000000000000..b056ec6d5327b --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FloatFFT_1D.java @@ -0,0 +1,156 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.udf.builtin.relational.tvf.fft; + +/** Computes an in-place 1D forward DFT for interleaved complex float data. */ +public final class FloatFFT_1D { + + private final int length; + + public FloatFFT_1D(long length) { + if (length < 1 || length > Integer.MAX_VALUE) { + throw new IllegalArgumentException("FFT length must be a positive int-sized value."); + } + this.length = (int) length; + } + + public void complexForward(float[] values) { + if (values.length < 2 * length) { + throw new IllegalArgumentException("Input array length must be at least 2 * FFT length."); + } + if (length == 1) { + return; + } + if (isPowerOfTwo(length)) { + transform(values, length, false); + } else { + bluesteinForward(values); + } + } + + private void bluesteinForward(float[] values) { + int convolutionLength = nextPowerOfTwo(2 * length - 1); + float[] a = new float[2 * convolutionLength]; + float[] b = new float[2 * convolutionLength]; + + for (int i = 0; i < length; i++) { + double angle = Math.PI * i * (double) i / length; + float cos = (float) Math.cos(angle); + float sin = (float) Math.sin(angle); + float real = values[2 * i]; + float imaginary = values[2 * i + 1]; + + a[2 * i] = real * cos + imaginary * sin; + a[2 * i + 1] = imaginary * cos - real * sin; + b[2 * i] = cos; + b[2 * i + 1] = sin; + if (i > 0) { + b[2 * (convolutionLength - i)] = cos; + b[2 * (convolutionLength - i) + 1] = sin; + } + } + + transform(a, convolutionLength, false); + transform(b, convolutionLength, false); + for (int i = 0; i < convolutionLength; i++) { + int offset = 2 * i; + float real = a[offset] * b[offset] - a[offset + 1] * b[offset + 1]; + float imaginary = a[offset] * b[offset + 1] + a[offset + 1] * b[offset]; + a[offset] = real; + a[offset + 1] = imaginary; + } + transform(a, convolutionLength, true); + + for (int i = 0; i < length; i++) { + double angle = Math.PI * i * (double) i / length; + float cos = (float) Math.cos(angle); + float sin = (float) Math.sin(angle); + float real = a[2 * i]; + float imaginary = a[2 * i + 1]; + values[2 * i] = real * cos + imaginary * sin; + values[2 * i + 1] = imaginary * cos - real * sin; + } + } + + private static void transform(float[] values, int size, boolean inverse) { + for (int i = 1, j = 0; i < size; i++) { + int bit = size >>> 1; + while ((j & bit) != 0) { + j ^= bit; + bit >>>= 1; + } + j ^= bit; + if (i < j) { + swap(values, 2 * i, 2 * j); + swap(values, 2 * i + 1, 2 * j + 1); + } + } + + for (int step = 2; step <= size; step <<= 1) { + double angle = (inverse ? 2.0 : -2.0) * Math.PI / step; + double stepReal = Math.cos(angle); + double stepImaginary = Math.sin(angle); + int halfStep = step >>> 1; + for (int block = 0; block < size; block += step) { + double factorReal = 1.0; + double factorImaginary = 0.0; + for (int j = 0; j < halfStep; j++) { + int even = 2 * (block + j); + int odd = 2 * (block + j + halfStep); + float oddReal = (float) (values[odd] * factorReal - values[odd + 1] * factorImaginary); + float oddImaginary = + (float) (values[odd] * factorImaginary + values[odd + 1] * factorReal); + float evenReal = values[even]; + float evenImaginary = values[even + 1]; + + values[even] = evenReal + oddReal; + values[even + 1] = evenImaginary + oddImaginary; + values[odd] = evenReal - oddReal; + values[odd + 1] = evenImaginary - oddImaginary; + + double nextFactorReal = factorReal * stepReal - factorImaginary * stepImaginary; + factorImaginary = factorReal * stepImaginary + factorImaginary * stepReal; + factorReal = nextFactorReal; + } + } + } + + if (inverse) { + for (int i = 0; i < 2 * size; i++) { + values[i] /= size; + } + } + } + + private static boolean isPowerOfTwo(int value) { + return (value & (value - 1)) == 0; + } + + private static int nextPowerOfTwo(int value) { + int highestOneBit = Integer.highestOneBit(value); + return value == highestOneBit ? value : highestOneBit << 1; + } + + private static void swap(float[] values, int left, int right) { + float tmp = values[left]; + values[left] = values[right]; + values[right] = tmp; + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunctionTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunctionTest.java new file mode 100644 index 0000000000000..6e00959b434c7 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FFTTableFunctionTest.java @@ -0,0 +1,504 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.udf.builtin.relational.tvf; + +import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils; +import org.apache.iotdb.udf.api.exception.UDFException; +import org.apache.iotdb.udf.api.relational.access.Record; +import org.apache.iotdb.udf.api.relational.table.TableFunctionAnalysis; +import org.apache.iotdb.udf.api.relational.table.TableFunctionHandle; +import org.apache.iotdb.udf.api.relational.table.argument.Argument; +import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument; +import org.apache.iotdb.udf.api.relational.table.argument.TableArgument; +import org.apache.iotdb.udf.api.relational.table.processor.TableFunctionDataProcessor; +import org.apache.iotdb.udf.api.type.Type; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.read.common.block.column.DoubleColumnBuilder; +import org.apache.tsfile.read.common.block.column.LongColumnBuilder; +import org.apache.tsfile.utils.Binary; +import org.junit.Test; + +import java.io.File; +import java.time.LocalDate; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +public class FFTTableFunctionTest { + + private static final double DELTA = 1e-9; + + private final FFTTableFunction function = new FFTTableFunction(); + + @Test + public void testWritesFullSpectrumAndZeroPadsToSpecifiedN() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(true, 4L); + processor.process(record(0L, 1.0), Collections.emptyList(), null); + + List 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 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 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); + processor.process(record(0L, 1.0), Collections.emptyList(), null); + processor.process(record(1L, 2.0), Collections.emptyList(), null); + + assertSemanticException( + () -> processor.process(nullValueRecord(2L), Collections.emptyList(), null), + "FFT does not support null values in column [value]."); + } + + @Test + public void testInfersSampleIntervalFromFullInputRangeBeyondSpecifiedN() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(false, 2L); + processor.process(record(0L, 1.0), Collections.emptyList(), null); + processor.process(record(1L, 2.0), Collections.emptyList(), null); + processor.process(record(3L, 3.0), Collections.emptyList(), null); + + List builders = createOutputBuilders(2); + processor.finish(builders, null); + + double intervalSeconds = + 1.5 * TimestampPrecisionUtils.currPrecision.toNanos(1L) / 1_000_000_000.0; + assertLongColumn(builders.get(0).build(), 0L, 1L); + assertDoubleColumn(builders.get(1).build(), 0.0, -1.0 / (2.0 * intervalSeconds)); + assertDoubleColumn(builders.get(2).build(), 3.0, -1.0); + assertDoubleColumn(builders.get(3).build(), 0.0, 0.0); + } + + @Test + public void testRejectsDuplicateTime() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(false); + processor.process(record(1L, 1.0), Collections.emptyList(), null); + + assertSemanticException( + () -> processor.process(record(1L, 2.0), Collections.emptyList(), null), + "The time column of FFT input must be strictly ascending within each partition."); + } + + @Test + public void testRejectsOutOfOrderTime() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(false); + processor.process(record(2L, 1.0), Collections.emptyList(), null); + + assertSemanticException( + () -> processor.process(record(1L, 2.0), Collections.emptyList(), null), + "The time column of FFT input must be strictly ascending within each partition."); + } + + @Test + public void testRejectsSingleRowWithoutSampleInterval() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(false); + processor.process(record(1L, 1.0), Collections.emptyList(), null); + + assertSemanticException( + () -> processor.finish(Collections.emptyList(), null), + "FFT requires at least two rows to infer SAMPLE_INTERVAL."); + } + + @Test + public void testInfersSampleIntervalFromPartitionTimeRange() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(false); + processor.process(record(0L, 1.0), Collections.emptyList(), null); + processor.process(record(1L, 2.0), Collections.emptyList(), null); + processor.process(record(3L, 3.0), Collections.emptyList(), null); + + List 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 testUsesExplicitSampleIntervalWithoutGapValidation() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(true); + processor.process(record(0L, 1.0), Collections.emptyList(), null); + processor.process(record(2L, 2.0), Collections.emptyList(), null); + + List builders = createOutputBuilders(2); + processor.finish(builders, null); + + double intervalSeconds = TimestampPrecisionUtils.currPrecision.toNanos(1L) / 1_000_000_000.0; + assertLongColumn(builders.get(0).build(), 0L, 1L); + assertDoubleColumn(builders.get(1).build(), 0.0, -1.0 / (2.0 * intervalSeconds)); + } + + @Test + public void testRejectsDefaultTransformLengthAboveLimit() throws UDFException { + TableFunctionDataProcessor processor = createProcessor(true); + for (long time = 0; time < 65_536L; time++) { + processor.process(record(time, 1.0), Collections.emptyList(), null); + } + + assertSemanticException( + () -> processor.process(record(65_536L, 1.0), Collections.emptyList(), null), + "FFT transform length N must not exceed 65536."); + } + + @Test + public void testRejectsDefaultSpectrumBufferAboveLimit() { + assertSemanticException( + () -> FFTTableFunction.validateTransformLength(65_536L, 129), + "FFT spectrum buffer is too large. Reduce N or the number of numeric columns."); + } + + @Test + public void testAnalyzeUsesSpecifiedTimeColumn() throws UDFException { + Map arguments = createArguments("event_time", "event_time"); + + TableFunctionAnalysis analysis = function.analyze(arguments); + + assertEquals( + Arrays.asList(0, 1), + analysis.getRequiredColumns().get(FFTTableFunction.DATA_PARAMETER_NAME)); + assertEquals( + "value_real", analysis.getProperColumnSchema().get().getFields().get(2).getName().get()); + assertEquals( + "value_imag", analysis.getProperColumnSchema().get().getFields().get(3).getName().get()); + } + + @Test + public void testAnalyzePreservesValueColumnNamesContainingComma() throws UDFException { + Map arguments = createArguments("time", "time", "value,with,comma"); + + TableFunctionAnalysis analysis = function.analyze(arguments); + + assertEquals( + "value,with,comma_real", + analysis.getProperColumnSchema().get().getFields().get(2).getName().get()); + assertEquals( + "value,with,comma_imag", + analysis.getProperColumnSchema().get().getFields().get(3).getName().get()); + + TableFunctionHandle handle = function.createTableFunctionHandle(); + handle.deserialize(analysis.getTableFunctionHandle().serialize()); + TableFunctionDataProcessor processor = function.getProcessorProvider(handle).getDataProcessor(); + processor.process(record(0L, 1.0), Collections.emptyList(), null); + assertSemanticException( + () -> processor.process(nullValueRecord(1L), Collections.emptyList(), null), + "FFT does not support null values in column [value,with,comma]."); + } + + @Test + public void testAnalyzeRejectsOrderByDifferentFromSpecifiedTimeColumn() { + assertSemanticException( + () -> { + try { + function.analyze(createArguments("event_time", "time")); + } catch (UDFException e) { + throw new AssertionError(e); + } + }, + "The ORDER BY clause of the DATA argument must contain exactly the time column specified by the TIMECOL argument."); + } + + private TableFunctionDataProcessor createProcessor(boolean sampleIntervalSpecified) + throws UDFException { + return createProcessor(sampleIntervalSpecified, -1L); + } + + private TableFunctionDataProcessor createProcessor( + boolean sampleIntervalSpecified, long transformLength) throws UDFException { + return createProcessor(Type.DOUBLE, sampleIntervalSpecified, transformLength); + } + + private TableFunctionDataProcessor createProcessor( + Type valueType, boolean sampleIntervalSpecified, long transformLength) throws UDFException { + Map arguments = new HashMap<>(); + arguments.put( + FFTTableFunction.DATA_PARAMETER_NAME, + new TableArgument( + Arrays.asList(Optional.of("time"), Optional.of("value")), + Arrays.asList(Type.TIMESTAMP, valueType), + Collections.emptyList(), + Collections.singletonList("time"), + false)); + arguments.put(FFTTableFunction.TIMECOL_PARAMETER_NAME, new ScalarArgument(Type.STRING, "time")); + arguments.put( + FFTTableFunction.SAMPLE_INTERVAL_PARAMETER_NAME, + new ScalarArgument(Type.INT64, sampleIntervalSpecified ? 1L : Long.MIN_VALUE)); + arguments.put( + FFTTableFunction.SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME, + new ScalarArgument(Type.BOOLEAN, sampleIntervalSpecified)); + arguments.put( + FFTTableFunction.N_PARAMETER_NAME, new ScalarArgument(Type.INT64, transformLength)); + arguments.put( + FFTTableFunction.NORM_PARAMETER_NAME, new ScalarArgument(Type.STRING, "backward")); + + return function + .getProcessorProvider(function.analyze(arguments).getTableFunctionHandle()) + .getDataProcessor(); + } + + private Map createArguments(String timeColumn, String orderByColumn) { + return createArguments(timeColumn, orderByColumn, "value"); + } + + private Map createArguments( + String timeColumn, String orderByColumn, String valueColumnName) { + Map arguments = new HashMap<>(); + arguments.put( + FFTTableFunction.DATA_PARAMETER_NAME, + new TableArgument( + Arrays.asList(Optional.of(timeColumn), Optional.of(valueColumnName)), + Arrays.asList(Type.TIMESTAMP, Type.DOUBLE), + Collections.emptyList(), + Collections.singletonList(orderByColumn), + false)); + arguments.put( + FFTTableFunction.TIMECOL_PARAMETER_NAME, new ScalarArgument(Type.STRING, timeColumn)); + arguments.put( + FFTTableFunction.SAMPLE_INTERVAL_PARAMETER_NAME, new ScalarArgument(Type.INT64, 1L)); + arguments.put( + FFTTableFunction.SAMPLE_INTERVAL_SPECIFIED_PARAMETER_NAME, + new ScalarArgument(Type.BOOLEAN, true)); + arguments.put(FFTTableFunction.N_PARAMETER_NAME, new ScalarArgument(Type.INT64, -1L)); + arguments.put( + FFTTableFunction.NORM_PARAMETER_NAME, new ScalarArgument(Type.STRING, "backward")); + return arguments; + } + + private Record record(long time, double value) { + return new SimpleRecord(time, value); + } + + private Record record(long time, float value) { + return new SimpleRecord(time, value); + } + + private Record nullValueRecord(long time) { + return new SimpleRecord(time, null); + } + + private List createOutputBuilders(int expectedPositionCount) { + return Arrays.asList( + new LongColumnBuilder(null, expectedPositionCount), + new DoubleColumnBuilder(null, expectedPositionCount), + new DoubleColumnBuilder(null, expectedPositionCount), + new DoubleColumnBuilder(null, expectedPositionCount)); + } + + private void assertLongColumn(Column column, long... expected) { + assertEquals(expected.length, column.getPositionCount()); + for (int i = 0; i < expected.length; i++) { + assertEquals(expected[i], column.getLong(i)); + } + } + + private void assertDoubleColumn(Column column, double... expected) { + assertDoubleColumnWithDelta(column, DELTA, expected); + } + + private void assertDoubleColumnWithDelta(Column column, double delta, double... expected) { + assertEquals(expected.length, column.getPositionCount()); + for (int i = 0; i < expected.length; i++) { + assertEquals(expected[i], column.getDouble(i), delta); + } + } + + private void assertSemanticException(Runnable runnable, String message) { + try { + runnable.run(); + fail(); + } catch (SemanticException e) { + assertEquals(message, e.getMessage()); + } + } + + private static class SimpleRecord implements Record { + private final long time; + private final Number value; + + private SimpleRecord(long time, Number value) { + this.time = time; + this.value = value; + } + + @Override + public int getInt(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public long getLong(int columnIndex) { + if (columnIndex == 0) { + return time; + } + throw new UnsupportedOperationException(); + } + + @Override + public float getFloat(int columnIndex) { + if (columnIndex == 1 && value instanceof Float) { + return value.floatValue(); + } + throw new UnsupportedOperationException(); + } + + @Override + public double getDouble(int columnIndex) { + if (columnIndex == 1 && value instanceof Double) { + return value.doubleValue(); + } + throw new UnsupportedOperationException(); + } + + @Override + public boolean getBoolean(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public Binary getBinary(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public String getString(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public LocalDate getLocalDate(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public Object getObject(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public Optional getObjectFile(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public long objectLength(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public Binary readObject(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public Binary readObject(int columnIndex, long offset, int length) { + throw new UnsupportedOperationException(); + } + + @Override + public Type getDataType(int columnIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean isNull(int columnIndex) { + if (columnIndex == 1) { + return value == null; + } + return false; + } + + @Override + public int size() { + return 2; + } + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FFT1DTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FFT1DTest.java new file mode 100644 index 0000000000000..86f9d6c11fa96 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/fft/FFT1DTest.java @@ -0,0 +1,109 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.udf.builtin.relational.tvf.fft; + +import org.junit.Test; + +import static org.junit.Assert.assertArrayEquals; + +public class FFT1DTest { + + @Test + public void testDoubleComplexForwardPowerOfTwoLength() { + double[] values = {1.0, 0.5, -2.0, 1.0, 0.0, -1.5, 3.0, 2.0}; + double[] expected = directDft(values); + + new DoubleFFT_1D(4).complexForward(values); + + assertArrayEquals(expected, values, 1e-9); + } + + @Test + public void testDoubleComplexForwardNonPowerOfTwoLength() { + double[] values = {1.0, 0.0, 2.0, -0.5, -1.0, 1.5, 0.0, 0.25, 3.0, -2.0}; + double[] expected = directDft(values); + + new DoubleFFT_1D(5).complexForward(values); + + assertArrayEquals(expected, values, 1e-9); + } + + @Test + public void testFloatComplexForwardPowerOfTwoLength() { + float[] values = {1.0f, 0.5f, -2.0f, 1.0f, 0.0f, -1.5f, 3.0f, 2.0f}; + float[] expected = directDft(values); + + new FloatFFT_1D(4).complexForward(values); + + assertArrayEquals(expected, values, 1e-5f); + } + + @Test + public void testFloatComplexForwardNonPowerOfTwoLength() { + float[] values = {1.0f, 0.0f, 2.0f, -0.5f, -1.0f, 1.5f, 0.0f, 0.25f, 3.0f, -2.0f}; + float[] expected = directDft(values); + + new FloatFFT_1D(5).complexForward(values); + + assertArrayEquals(expected, values, 1e-5f); + } + + private static double[] directDft(double[] values) { + int length = values.length / 2; + double[] result = new double[values.length]; + for (int frequencyIndex = 0; frequencyIndex < length; frequencyIndex++) { + double real = 0.0; + double imaginary = 0.0; + for (int timeIndex = 0; timeIndex < length; timeIndex++) { + double angle = -2.0 * Math.PI * frequencyIndex * timeIndex / length; + double cos = Math.cos(angle); + double sin = Math.sin(angle); + double inputReal = values[2 * timeIndex]; + double inputImaginary = values[2 * timeIndex + 1]; + real += inputReal * cos - inputImaginary * sin; + imaginary += inputReal * sin + inputImaginary * cos; + } + result[2 * frequencyIndex] = real; + result[2 * frequencyIndex + 1] = imaginary; + } + return result; + } + + private static float[] directDft(float[] values) { + int length = values.length / 2; + float[] result = new float[values.length]; + for (int frequencyIndex = 0; frequencyIndex < length; frequencyIndex++) { + double real = 0.0; + double imaginary = 0.0; + for (int timeIndex = 0; timeIndex < length; timeIndex++) { + double angle = -2.0 * Math.PI * frequencyIndex * timeIndex / length; + double cos = Math.cos(angle); + double sin = Math.sin(angle); + double inputReal = values[2 * timeIndex]; + double inputImaginary = values[2 * timeIndex + 1]; + real += inputReal * cos - inputImaginary * sin; + imaginary += inputReal * sin + inputImaginary * cos; + } + result[2 * frequencyIndex] = (float) real; + result[2 * frequencyIndex + 1] = (float) imaginary; + } + return result; + } +}