This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 6a236ff51d6 Implementation of LowPass, HighPass, and XCorr Table
Functions (#18539)
6a236ff51d6 is described below
commit 6a236ff51d6abd54aaf03fdc8f2b37179dfc5f3e
Author: CYB <[email protected]>
AuthorDate: Wed Sep 2 15:30:15 2026 +0800
Implementation of LowPass, HighPass, and XCorr Table Functions (#18539)
---
.../relational/it/db/it/IoTDBWindowTVFIT.java | 273 ++++++++++++++++++
.../relational/analyzer/StatementAnalyzer.java | 3 +
.../plan/relational/planner/RelationPlanner.java | 15 +-
.../apache/iotdb/commons/i18n/CommonMessages.java | 255 ++++++++++++-----
.../apache/iotdb/commons/i18n/CommonMessages.java | 9 +-
.../function/TableBuiltinTableFunction.java | 14 +-
.../tvf/FilterTransferTableFunction.java | 315 +++++++++++++++++++++
.../relational/tvf/HighPassTableFunction.java | 81 ++++++
.../relational/tvf/LowPassTableFunction.java | 77 +++++
.../builtin/relational/tvf/M4TableFunction.java | 87 +-----
.../udf/builtin/relational/tvf/WindowTVFUtils.java | 185 ++++++++++++
.../builtin/relational/tvf/XCorrTableFunction.java | 249 ++++++++++++++++
.../builtin/relational/tvf/fft/DoubleFFT_1D.java | 34 ++-
.../udf/builtin/relational/tvf/fft/FFT1DTest.java | 39 +++
14 files changed, 1476 insertions(+), 160 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java
index 3aa33889568..a225039b474 100644
---
a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java
@@ -31,17 +31,23 @@ 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 java.sql.Types;
import java.util.Arrays;
import java.util.List;
import static org.apache.iotdb.db.it.utils.TestUtils.*;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
@RunWith(IoTDBTestRunner.class)
@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
public class IoTDBWindowTVFIT {
private static final String DATABASE_NAME = "test";
+ private static final double DELTA = 1e-9;
private static final String[] sqls =
new String[] {
"CREATE DATABASE " + DATABASE_NAME,
@@ -1347,4 +1353,271 @@ public class IoTDBWindowTVFIT {
"701: The type of the column [s1] is not as expected.",
DATABASE_NAME);
}
+
+ @Test
+ public void testLowPassWithOnePartition() {
+ String[] expectedHeader = new String[] {"stock_id", "time",
"lowpass(price)", "lowpass(s1)"};
+ String[] retArray =
+ new String[] {
+ "AAPL,2021-01-01T09:05:00.000Z,101.66666666666667,101.0,",
+ "AAPL,2021-01-01T09:07:00.000Z,101.66666666666667,101.0,",
+ "AAPL,2021-01-01T09:09:00.000Z,101.66666666666667,101.0,",
+ "TESL,2021-01-01T09:06:00.000Z,199.0,212.0,",
+ "TESL,2021-01-01T09:07:00.000Z,199.0,212.0,",
+ "TESL,2021-01-01T09:15:00.000Z,199.0,212.0,"
+ };
+ tableResultSetEqualWithTolerance(
+ "SELECT * FROM LOWPASS(DATA => bid PARTITION BY stock_id ORDER BY
time, "
+ + "TIMECOL => 'time', WPASS => 0.5) ORDER BY stock_id, time",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testLowPassWithNullValues() {
+ String[] expectedHeader = new String[] {"device_id", "time",
"lowpass(s2)"};
+ String[] retArray =
+ new String[] {
+ "device_1,1970-01-01T00:00:00.001Z,23.5,",
+ "device_1,1970-01-01T00:00:00.003Z,null,",
+ "device_1,1970-01-01T00:00:00.006Z,null,",
+ "device_1,1970-01-01T00:00:00.009Z,null,",
+ "device_1,1970-01-01T00:00:00.020Z,23.5,"
+ };
+ tableResultSetEqualWithTolerance(
+ "SELECT * FROM LOWPASS(DATA => "
+ + "(SELECT time, device_id, s2 FROM table1 WHERE device_id =
'device_1') "
+ + "PARTITION BY device_id ORDER BY time, TIMECOL => 'time', WPASS
=> 0.5) "
+ + "ORDER BY time",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testLowPassWithMultiplePartitionColumns() {
+ String[] expectedHeader = new String[] {"factory_id", "device_id", "time",
"lowpass(s1)"};
+ String[] retArray =
+ new String[] {
+ "F1,device_1,1970-01-01T00:00:00.001Z,10.0,",
+ "F1,device_1,1970-01-01T00:00:00.005Z,10.0,",
+ "F1,device_1,1970-01-01T00:00:00.009Z,10.0,",
+ "F1,device_2,1970-01-01T00:00:00.002Z,22.5,",
+ "F1,device_2,1970-01-01T00:00:00.011Z,22.5,",
+ "F2,device_1,1970-01-01T00:00:00.003Z,30.0,"
+ };
+ tableResultSetEqualWithTolerance(
+ "SELECT * FROM LOWPASS(DATA => table5 "
+ + "PARTITION BY (factory_id, device_id) ORDER BY time, "
+ + "TIMECOL => 'time', WPASS => 0.5) ORDER BY factory_id,
device_id, time",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testLowPassRejectsUnsupportedCalculationColumnType() {
+ tableAssertTestFail(
+ "SELECT * FROM LOWPASS(DATA => (SELECT time, device_id, s3 FROM
table1) "
+ + "PARTITION BY device_id ORDER BY time, TIMECOL => 'time', WPASS
=> 0.5)",
+ "701: Only column with double, float, int32, int64 can be calculated
by the function, s3 is the STRING.",
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testHighWithOnePartition() {
+ String[] expectedHeader = new String[] {"stock_id", "time",
"highpass(price)", "highpass(s1)"};
+ String[] retArray =
+ new String[] {
+ "AAPL,2021-01-01T09:05:00.000Z,-1.6666666666666667,0.0,",
+ "AAPL,2021-01-01T09:07:00.000Z,1.3333333333333333,0.0,",
+ "AAPL,2021-01-01T09:09:00.000Z,0.3333333333333333,0.0,",
+ "TESL,2021-01-01T09:06:00.000Z,1.0,-110.0,",
+ "TESL,2021-01-01T09:07:00.000Z,3.0,-10.0,",
+ "TESL,2021-01-01T09:15:00.000Z,-4.0,120.0,"
+ };
+ tableResultSetEqualWithTolerance(
+ "SELECT * FROM HIGHPASS(DATA => bid PARTITION BY stock_id ORDER BY
time, "
+ + "TIMECOL => 'time', WPASS => 0.5) ORDER BY stock_id, time",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testHighPassWithNullValues() {
+ String[] expectedHeader = new String[] {"device_id", "time",
"highpass(s2)"};
+ String[] retArray =
+ new String[] {
+ "device_1,1970-01-01T00:00:00.001Z,-11.5,",
+ "device_1,1970-01-01T00:00:00.003Z,null,",
+ "device_1,1970-01-01T00:00:00.006Z,null,",
+ "device_1,1970-01-01T00:00:00.009Z,null,",
+ "device_1,1970-01-01T00:00:00.020Z,11.5,"
+ };
+ tableResultSetEqualWithTolerance(
+ "SELECT * FROM HIGHPASS(DATA => "
+ + "(SELECT time, device_id, s2 FROM table1 WHERE device_id =
'device_1') "
+ + "PARTITION BY device_id ORDER BY time, TIMECOL => 'time', WPASS
=> 0.5) "
+ + "ORDER BY time",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testHighPassWithMultiplePartitionColumns() {
+ String[] expectedHeader = new String[] {"factory_id", "device_id", "time",
"highpass(s1)"};
+ String[] retArray =
+ new String[] {
+ "F1,device_1,1970-01-01T00:00:00.001Z,0.0,",
+ "F1,device_1,1970-01-01T00:00:00.005Z,5.0,",
+ "F1,device_1,1970-01-01T00:00:00.009Z,-5.0,",
+ "F1,device_2,1970-01-01T00:00:00.002Z,-2.5,",
+ "F1,device_2,1970-01-01T00:00:00.011Z,2.5,",
+ "F2,device_1,1970-01-01T00:00:00.003Z,0.0,"
+ };
+ tableResultSetEqualWithTolerance(
+ "SELECT * FROM HIGHPASS(DATA => table5 "
+ + "PARTITION BY (factory_id, device_id) ORDER BY time, "
+ + "TIMECOL => 'time', WPASS => 0.5) ORDER BY factory_id,
device_id, time",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testHighPassRejectsUnsupportedCalculationColumnType() {
+ tableAssertTestFail(
+ "SELECT * FROM HIGHPASS(DATA => (SELECT time, device_id, s3 FROM
table1) "
+ + "PARTITION BY device_id ORDER BY time, TIMECOL => 'time', WPASS
=> 0.5)",
+ "701: Only column with double, float, int32, int64 can be calculated
by the function, s3 is the STRING.",
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testXCorrP0() {
+ String[] expectedHeader = new String[] {"stock_id", "xcorr(price, s1)"};
+ String[] retArray =
+ new String[] {
+ "AAPL,10100.0,",
+ "AAPL,10251.5,",
+ "AAPL,10268.333333333334,",
+ "AAPL,10352.5,",
+ "AAPL,10302.0,",
+ "TESL,66400.0,",
+ "TESL,53732.0,",
+ "TESL,41981.333333333336,",
+ "TESL,29997.0,",
+ "TESL,19890.0,"
+ };
+ tableResultSetEqualTest(
+ "SELECT * FROM XCORR(DATA => bid PARTITION BY stock_id ORDER BY time, "
+ + "TIMECOL => 'time') ORDER BY stock_id",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testXCorrWithNullValues() {
+ String[] expectedHeader = new String[] {"device_id", "xcorr(s1, s2)"};
+ String[] retArray =
+ new String[] {
+ "device_1,525.0,",
+ "device_1,175.0,",
+ "device_1,1050.0,",
+ "device_1,350.0,",
+ "device_1,790.0,",
+ "device_1,60.0,",
+ "device_1,360.0,",
+ "device_1,120.0,",
+ "device_1,480.0,"
+ };
+ tableResultSetEqualTest(
+ "SELECT * FROM XCORR(DATA => "
+ + "(SELECT time, device_id, s1, s2 FROM table1 WHERE device_id =
'device_1') "
+ + "PARTITION BY device_id ORDER BY time, TIMECOL => 'time')",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testXCorrWithMultiplePartitionColumns() {
+ String[] expectedHeader = new String[] {"factory_id", "device_id",
"xcorr(s1, s2)"};
+ String[] retArray =
+ new String[] {
+ "F1,device_1,50.0,",
+ "F1,device_1,112.5,",
+ "F1,device_1,116.66666666666667,",
+ "F1,device_1,112.5,",
+ "F1,device_1,50.0,",
+ "F1,device_2,500.0,",
+ "F1,device_2,512.5,",
+ "F1,device_2,500.0,",
+ "F2,device_1,900.0,"
+ };
+ tableResultSetEqualTest(
+ "SELECT * FROM XCORR(DATA => "
+ + "(SELECT time, factory_id, device_id, s1, s1 AS s2 FROM table5) "
+ + "PARTITION BY (factory_id, device_id) ORDER BY time, TIMECOL =>
'time') "
+ + "ORDER BY factory_id, device_id",
+ expectedHeader,
+ retArray,
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testXCorrRejectsUnsupportedCalculationColumnType() {
+ tableAssertTestFail(
+ "SELECT * FROM XCORR(DATA => (SELECT time, device_id, s1, s3 FROM
table1) "
+ + "PARTITION BY device_id ORDER BY time, TIMECOL => 'time')",
+ "701: Only column with double, float, int32, int64 can be calculated
by the function, s3 is the STRING.",
+ DATABASE_NAME);
+ }
+
+ @Test
+ public void testXCorrRejectsUnexpectedCalculationColumnCount() {
+ tableAssertTestFail(
+ "SELECT * FROM XCORR(DATA => (select time, device_id, int_val,
long_val, float_val from multi_type) PARTITION BY device_id ORDER BY time,
TIMECOL => 'time')",
+ "701: XCorr requires exactly two calculation columns, but found 3.",
+ DATABASE_NAME);
+ }
+
+ private static void tableResultSetEqualWithTolerance(
+ String sql, String[] expectedHeader, String[] expectedRetArray, String
database) {
+ try (Connection connection = EnvFactory.getEnv().getTableConnection();
+ Statement statement = connection.createStatement()) {
+ connection.setClientInfo("time_zone", "+00:00");
+ statement.execute("USE " + database);
+ try (ResultSet resultSet = statement.executeQuery(sql)) {
+ ResultSetMetaData metaData = resultSet.getMetaData();
+ assertEquals(expectedHeader.length, metaData.getColumnCount());
+ for (int i = 1; i <= metaData.getColumnCount(); i++) {
+ assertEquals(expectedHeader[i - 1], metaData.getColumnName(i));
+ }
+
+ int rowIndex = 0;
+ while (resultSet.next()) {
+ String[] expectedColumns = expectedRetArray[rowIndex].split(",", -1);
+ for (int i = 1; i <= expectedHeader.length; i++) {
+ String expected = expectedColumns[i - 1];
+ if (resultSet.getString(i) == null) {
+ assertEquals("null", expected);
+ } else if (metaData.getColumnType(i) == Types.DOUBLE) {
+ assertEquals(Double.parseDouble(expected),
resultSet.getDouble(i), DELTA);
+ } else {
+ assertEquals(expected, resultSet.getString(i));
+ }
+ }
+ rowIndex++;
+ }
+ assertEquals(expectedRetArray.length, rowIndex);
+ }
+ } catch (SQLException e) {
+ fail(e.getMessage());
+ }
+ }
}
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 9a54519953d..b5c394539f2 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
@@ -5923,6 +5923,9 @@ public class StatementAnalyzer {
private boolean isPartitionColumnsProvidedByProperSchema(String
functionName) {
return
TableBuiltinTableFunction.M4.getFunctionName().equalsIgnoreCase(functionName)
+ ||
TableBuiltinTableFunction.LOWPASS.getFunctionName().equalsIgnoreCase(functionName)
+ ||
TableBuiltinTableFunction.HIGHPASS.getFunctionName().equalsIgnoreCase(functionName)
+ ||
TableBuiltinTableFunction.XCORR.getFunctionName().equalsIgnoreCase(functionName)
||
TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName);
}
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 7fc367b3785..6f5f3c278ed 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
@@ -1572,12 +1572,7 @@ public class RelationPlanner implements
AstVisitor<RelationPlan, Void> {
symbol ->
new TableFunctionNode.PassThroughColumn(symbol,
partitionBy.contains(symbol)))
.forEach(passThroughColumns::add);
- } else if (!TableBuiltinTableFunction.M4
- .getFunctionName()
- .equalsIgnoreCase(functionAnalysis.getFunctionName())
- && !TableBuiltinTableFunction.FFT
- .getFunctionName()
- .equalsIgnoreCase(functionAnalysis.getFunctionName())
+ } else if (needAddPartitionColumn(functionAnalysis.getFunctionName())
&& tableArgument.getPartitionBy().isPresent()) {
tableArgument.getPartitionBy().get().stream()
// the original symbols for partitioning columns, not coerced
@@ -1613,6 +1608,14 @@ public class RelationPlanner implements
AstVisitor<RelationPlan, Void> {
return new RelationPlan(root, analysis.getScope(node),
outputSymbols.build(), outerContext);
}
+ private boolean needAddPartitionColumn(String functionName) {
+ return
!TableBuiltinTableFunction.M4.getFunctionName().equalsIgnoreCase(functionName)
+ &&
!TableBuiltinTableFunction.LOWPASS.getFunctionName().equalsIgnoreCase(functionName)
+ &&
!TableBuiltinTableFunction.HIGHPASS.getFunctionName().equalsIgnoreCase(functionName)
+ &&
!TableBuiltinTableFunction.XCORR.getFunctionName().equalsIgnoreCase(functionName)
+ &&
!TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName);
+ }
+
private RelationPlan planExternalTsFileScan(
TableFunctionInvocation node, TableFunctionInvocationAnalysis
functionAnalysis) {
if (!(functionAnalysis.getTableFunctionHandle()
diff --git
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
index 94194a086b4..40c75066d87 100644
---
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
@@ -38,8 +38,7 @@ public final class CommonMessages {
public static final String UNKNOWN_NODE_STATUS = "Unknown NodeStatus %s.";
// --- consensus ---
- public static final String UNRECOGNIZED_CONSENSUS_GROUP_ID =
- "Unrecognized ConsensusGroupId: %s";
+ public static final String UNRECOGNIZED_CONSENSUS_GROUP_ID = "Unrecognized
ConsensusGroupId: %s";
public static final String IOTV2_BG_NOT_TERMINATED =
"IoTV2 background service did not terminate within {}s";
public static final String IOTV2_BG_STILL_RUNNING =
@@ -136,93 +135,197 @@ public final class CommonMessages {
public static final String MAP_MUST_NOT_BE_NULL = "Map must not be null.";
public static final String MAP_ENTRY_MUST_NOT_BE_NULL = "Map Entry must not
be null.";
public static final String ITERATOR_MUST_NOT_BE_NULL = "Iterator must not be
null";
- public static final String ITERATOR_REMOVE_ONLY_AFTER_NEXT = "Iterator
remove() can only be called once after next()";
+ public static final String ITERATOR_REMOVE_ONLY_AFTER_NEXT =
+ "Iterator remove() can only be called once after next()";
public static final String FAIL_TO_GET_DATA_TYPE_IN_ROW = "Fail to get data
type in row ";
//
---------------------------------------------------------------------------
// Additional auto-collected messages
//
---------------------------------------------------------------------------
- public static final String
LOG_STEP_METRICS_ARG_ARG_TOTAL_ARG_SUM_2FMS_AVG_ARG_87491AB0 = "step metrics
[%d]-[%s] - Total: %d, SUM: %.2fms, AVG: %fms, Last%dAVG: %fms";
- public static final String
LOG_ERROR_OCCURRED_DURING_TRANSFERRING_FILE_ARG_BYTEBUFFER_CAUSE_ARG_FEDC38A3 =
"Error occurred during transferring file{} to ByteBuffer, the cause is {}";
- public static final String
LOG_ERROR_OCCURRED_DURING_WRITING_BYTEBUFFER_ARG_CAUSE_ARG_F3AD2DA0 = "Error
occurred during writing bytebuffer to {} , the cause is {}";
- public static final String EXCEPTION_SIZE_FILE_EXCEED_ARG_BYTES_C60F1149 =
"Size of file exceed %d bytes";
- public static final String
EXCEPTION_UNRECOGNIZED_TCONSENSUSGROUPTYPE_9204FF8E = "Unrecognized
TConsensusGroupType: ";
+ public static final String
LOG_STEP_METRICS_ARG_ARG_TOTAL_ARG_SUM_2FMS_AVG_ARG_87491AB0 =
+ "step metrics [%d]-[%s] - Total: %d, SUM: %.2fms, AVG: %fms, Last%dAVG:
%fms";
+ public static final String
+
LOG_ERROR_OCCURRED_DURING_TRANSFERRING_FILE_ARG_BYTEBUFFER_CAUSE_ARG_FEDC38A3 =
+ "Error occurred during transferring file{} to ByteBuffer, the cause
is {}";
+ public static final String
LOG_ERROR_OCCURRED_DURING_WRITING_BYTEBUFFER_ARG_CAUSE_ARG_F3AD2DA0 =
+ "Error occurred during writing bytebuffer to {} , the cause is {}";
+ public static final String EXCEPTION_SIZE_FILE_EXCEED_ARG_BYTES_C60F1149 =
+ "Size of file exceed %d bytes";
+ public static final String
EXCEPTION_UNRECOGNIZED_TCONSENSUSGROUPTYPE_9204FF8E =
+ "Unrecognized TConsensusGroupType: ";
public static final String EXCEPTION_ID_1F238F51 = " with id = ";
- public static final String
LOG_MEMORY_COST_RELEASED_LARGER_THAN_MEMORY_COST_MEMORY_BLOCK_ARG_00DD9DA9 =
"The memory cost to be released is larger than the memory cost of memory block
{}";
- public static final String
LOG_EXACTALLOCATEIFSUFFICIENT_FAILED_ALLOCATE_MEMORY_A47897D9 =
"exactAllocateIfSufficient: failed to allocate memory, ";
- public static final String
LOG_TOTAL_MEMORY_SIZE_ARG_BYTES_USED_MEMORY_SIZE_ARG_BYTES_5FB5059F = "total
memory size {} bytes, used memory size {} bytes, ";
- public static final String
LOG_REQUESTED_MEMORY_SIZE_ARG_BYTES_USED_THRESHOLD_ARG_D7061DEB = "requested
memory size {} bytes, used threshold {}";
- public static final String LOG_TRYALLOCATE_ALLOCATED_MEMORY_B3D564D9 =
"tryAllocate: allocated memory, ";
- public static final String
LOG_ORIGINAL_REQUESTED_MEMORY_SIZE_ARG_BYTES_03D28A6B = "original requested
memory size {} bytes, ";
- public static final String
LOG_ACTUAL_REQUESTED_MEMORY_SIZE_ARG_BYTES_62760058 = "actual requested memory
size {} bytes";
- public static final String LOG_TRYALLOCATE_FAILED_ALLOCATE_MEMORY_838FA6FB =
"tryAllocate: failed to allocate memory, ";
- public static final String LOG_REQUESTED_MEMORY_SIZE_ARG_BYTES_BF9CEF81 =
"requested memory size {} bytes";
- public static final String
LOG_GETORREGISTERMEMORYBLOCK_FAILED_MEMORY_BLOCK_ARG_ALREADY_EXISTS_42CA8914 =
"getOrRegisterMemoryBlock failed: memory block {} already exists, ";
- public static final String LOG_IT_S_SIZE_ARG_REQUESTED_SIZE_ARG_AF8F04B2 =
"it's size is {}, requested size is {}";
- public static final String
LOG_GETMEMORYMANAGER_MEMORY_MANAGER_ARG_ALREADY_EXISTS_IT_S_SIZE_ARG_0102560A =
"getMemoryManager: memory manager {} already exists, it's size is {}, enabled
is {}";
- public static final String
LOG_GETORCREATEMEMORYMANAGER_FAILED_TOTAL_MEMORY_SIZE_ARG_BYTES_LESS_THAN_ALLOCATED_3D110256
=
- "getOrCreateMemoryManager failed: total memory size {} bytes is less
than allocated memory"
- + " size {} bytes";
- public static final String
EXCEPTION_EXACTALLOCATE_FAILED_ALLOCATE_MEMORY_AFTER_ARG_RETRIES_957A647B =
"exactAllocate: failed to allocate memory after %d retries, ";
- public static final String
EXCEPTION_TOTAL_MEMORY_SIZE_ARG_BYTES_USED_MEMORY_SIZE_ARG_BYTES_9FC9A9C6 =
"total memory size %d bytes, used memory size %d bytes, ";
- public static final String
EXCEPTION_REQUESTED_MEMORY_SIZE_ARG_BYTES_E6340842 = "requested memory size %d
bytes";
- public static final String
EXCEPTION_REGISTER_MEMORY_BLOCK_ARG_FAILED_SIZEINBYTES_SHOULD_NON_NEGATIVE_EC54AA75
= "register memory block %s failed: sizeInBytes should be non-negative";
- public static final String
LOG_DELETE_SYSTEM_PROPERTIES_TMP_FILE_FAIL_YOU_MAY_MANUALLY_DELETE_F81C4A53 =
"Delete system.properties tmp file fail, you may manually delete it: {}";
- public static final String
LOG_FAILED_DELETE_SYSTEM_PROPERTIES_FILE_YOU_SHOULD_MANUALLY_DELETE_THEM_77F91A98
= "Failed to delete system.properties file, you should manually delete them:
{}, {}";
- public static final String
EXCEPTION_LENGTH_PARAMETERS_SHOULD_EVENLY_DIVIDED_2_BUT_ACTUAL_LENGTH_E9A792D9
= "Length of parameters should be evenly divided by 2, but the actual length is
";
- public static final String
EXCEPTION_TMP_SYSTEM_PROPERTIES_FILE_MUST_EXIST_CALL_REPLACEFORMALFILE_FA63B976
= "Tmp system properties file must exist when call replaceFormalFile";
- public static final String
LOG_UNRECOVERABLE_ERROR_OCCURS_CHANGE_SYSTEM_STATUS_READ_ONLY_BECAUSE_HANDLE_05C9AD1A
=
- "Unrecoverable error occurs! Change system status to read-only because
handle_system_error is"
- + " CHANGE_TO_READ_ONLY. Only query statements are permitted!";
- public static final String
LOG_UNRECOVERABLE_ERROR_OCCURS_SHUTDOWN_SYSTEM_DIRECTLY_BECAUSE_HANDLE_SYSTEM_ERROR_14FC06C9
= "Unrecoverable error occurs! Shutdown system directly because
handle_system_error is SHUTDOWN.";
- public static final String
EXCEPTION_TYPE_ARG_NOT_SUPPORTED_PIPE_RATE_AVERAGE_F74694AD = "The type %s is
not supported in pipe rate average.";
+ public static final String
+
LOG_MEMORY_COST_RELEASED_LARGER_THAN_MEMORY_COST_MEMORY_BLOCK_ARG_00DD9DA9 =
+ "The memory cost to be released is larger than the memory cost of
memory block {}";
+ public static final String
LOG_EXACTALLOCATEIFSUFFICIENT_FAILED_ALLOCATE_MEMORY_A47897D9 =
+ "exactAllocateIfSufficient: failed to allocate memory, ";
+ public static final String
LOG_TOTAL_MEMORY_SIZE_ARG_BYTES_USED_MEMORY_SIZE_ARG_BYTES_5FB5059F =
+ "total memory size {} bytes, used memory size {} bytes, ";
+ public static final String
LOG_REQUESTED_MEMORY_SIZE_ARG_BYTES_USED_THRESHOLD_ARG_D7061DEB =
+ "requested memory size {} bytes, used threshold {}";
+ public static final String LOG_TRYALLOCATE_ALLOCATED_MEMORY_B3D564D9 =
+ "tryAllocate: allocated memory, ";
+ public static final String
LOG_ORIGINAL_REQUESTED_MEMORY_SIZE_ARG_BYTES_03D28A6B =
+ "original requested memory size {} bytes, ";
+ public static final String
LOG_ACTUAL_REQUESTED_MEMORY_SIZE_ARG_BYTES_62760058 =
+ "actual requested memory size {} bytes";
+ public static final String LOG_TRYALLOCATE_FAILED_ALLOCATE_MEMORY_838FA6FB =
+ "tryAllocate: failed to allocate memory, ";
+ public static final String LOG_REQUESTED_MEMORY_SIZE_ARG_BYTES_BF9CEF81 =
+ "requested memory size {} bytes";
+ public static final String
+
LOG_GETORREGISTERMEMORYBLOCK_FAILED_MEMORY_BLOCK_ARG_ALREADY_EXISTS_42CA8914 =
+ "getOrRegisterMemoryBlock failed: memory block {} already exists, ";
+ public static final String LOG_IT_S_SIZE_ARG_REQUESTED_SIZE_ARG_AF8F04B2 =
+ "it's size is {}, requested size is {}";
+ public static final String
+
LOG_GETMEMORYMANAGER_MEMORY_MANAGER_ARG_ALREADY_EXISTS_IT_S_SIZE_ARG_0102560A =
+ "getMemoryManager: memory manager {} already exists, it's size is
{}, enabled is {}";
+ public static final String
+
LOG_GETORCREATEMEMORYMANAGER_FAILED_TOTAL_MEMORY_SIZE_ARG_BYTES_LESS_THAN_ALLOCATED_3D110256
=
+ "getOrCreateMemoryManager failed: total memory size {} bytes is less
than allocated memory"
+ + " size {} bytes";
+ public static final String
+
EXCEPTION_EXACTALLOCATE_FAILED_ALLOCATE_MEMORY_AFTER_ARG_RETRIES_957A647B =
+ "exactAllocate: failed to allocate memory after %d retries, ";
+ public static final String
+
EXCEPTION_TOTAL_MEMORY_SIZE_ARG_BYTES_USED_MEMORY_SIZE_ARG_BYTES_9FC9A9C6 =
+ "total memory size %d bytes, used memory size %d bytes, ";
+ public static final String
EXCEPTION_REQUESTED_MEMORY_SIZE_ARG_BYTES_E6340842 =
+ "requested memory size %d bytes";
+ public static final String
+
EXCEPTION_REGISTER_MEMORY_BLOCK_ARG_FAILED_SIZEINBYTES_SHOULD_NON_NEGATIVE_EC54AA75
=
+ "register memory block %s failed: sizeInBytes should be
non-negative";
+ public static final String
+
LOG_DELETE_SYSTEM_PROPERTIES_TMP_FILE_FAIL_YOU_MAY_MANUALLY_DELETE_F81C4A53 =
+ "Delete system.properties tmp file fail, you may manually delete it:
{}";
+ public static final String
+
LOG_FAILED_DELETE_SYSTEM_PROPERTIES_FILE_YOU_SHOULD_MANUALLY_DELETE_THEM_77F91A98
=
+ "Failed to delete system.properties file, you should manually delete
them: {}, {}";
+ public static final String
+
EXCEPTION_LENGTH_PARAMETERS_SHOULD_EVENLY_DIVIDED_2_BUT_ACTUAL_LENGTH_E9A792D9 =
+ "Length of parameters should be evenly divided by 2, but the actual
length is ";
+ public static final String
+
EXCEPTION_TMP_SYSTEM_PROPERTIES_FILE_MUST_EXIST_CALL_REPLACEFORMALFILE_FA63B976
=
+ "Tmp system properties file must exist when call replaceFormalFile";
+ public static final String
+
LOG_UNRECOVERABLE_ERROR_OCCURS_CHANGE_SYSTEM_STATUS_READ_ONLY_BECAUSE_HANDLE_05C9AD1A
=
+ "Unrecoverable error occurs! Change system status to read-only
because handle_system_error is"
+ + " CHANGE_TO_READ_ONLY. Only query statements are permitted!";
+ public static final String
+
LOG_UNRECOVERABLE_ERROR_OCCURS_SHUTDOWN_SYSTEM_DIRECTLY_BECAUSE_HANDLE_SYSTEM_ERROR_14FC06C9
=
+ "Unrecoverable error occurs! Shutdown system directly because
handle_system_error is SHUTDOWN.";
+ public static final String
EXCEPTION_TYPE_ARG_NOT_SUPPORTED_PIPE_RATE_AVERAGE_F74694AD =
+ "The type %s is not supported in pipe rate average.";
public static final String EXCEPTION_UNKNOWN_UDFTYPE_9A8D1B23 = "Unknown
UDFType:";
public static final String EXCEPTION_8S_5F5F831F = "%8s";
- public static final String EXCEPTION_CAN_NOT_RECOGNIZE_PIPETYPE_ARG_8850A249
= "Can not recognize PipeType %s.";
- public static final String
EXCEPTION_TARGETREGIONLIST_EMPTY_DEVICE_ARG_TIMESLOT_ARG_E7E5818C =
"targetRegionList is empty. device: %s, timeSlot: %s";
+ public static final String EXCEPTION_CAN_NOT_RECOGNIZE_PIPETYPE_ARG_8850A249
=
+ "Can not recognize PipeType %s.";
+ public static final String
EXCEPTION_TARGETREGIONLIST_EMPTY_DEVICE_ARG_TIMESLOT_ARG_E7E5818C =
+ "targetRegionList is empty. device: %s, timeSlot: %s";
public static final String EXCEPTION_DATABASE_18F8303F = "Database ";
- public static final String
EXCEPTION_NOT_EXISTS_FAILED_CREATE_AUTOMATICALLY_BECAUSE_ENABLE_AUTO_CREATE_SCHEMA_80DE1A4B
= " not exists and failed to create automatically because
enable_auto_create_schema is FALSE.";
+ public static final String
+
EXCEPTION_NOT_EXISTS_FAILED_CREATE_AUTOMATICALLY_BECAUSE_ENABLE_AUTO_CREATE_SCHEMA_80DE1A4B
=
+ " not exists and failed to create automatically because
enable_auto_create_schema is FALSE.";
public static final String EXCEPTION_PATH_DOES_NOT_EXIST_737CB95D = "Path
does not exist. ";
- public static final String
EXCEPTION_CAN_T_GET_NEXT_FOLDER_ARG_BECAUSE_THEY_ALL_FULL_A105BB2D = "Can't get
next folder from [%s], because they are all full.";
- public static final String
EXCEPTION_PARAMETER_ARG_CAN_NOT_ARG_PLEASE_SET_ARG_BECAUSE_ARG_749738D1 =
"Parameter %s can not be %s, please set to: %s. Because %s";
- public static final String EXCEPTION_QUERY_EXECUTION_TIME_OUT_A5DC7BFB =
"Query execution is time out";
- public static final String EXCEPTION_OBJECT_FILE_ARG_DOES_NOT_EXIST_7EA8CB1C
= "Object file %s does not exist";
- public static final String EXCEPTION_ARG_NOT_LEGAL_PRIVILEGE_504838E8 = "%s
is not a legal privilege";
+ public static final String
EXCEPTION_CAN_T_GET_NEXT_FOLDER_ARG_BECAUSE_THEY_ALL_FULL_A105BB2D =
+ "Can't get next folder from [%s], because they are all full.";
+ public static final String
+ EXCEPTION_PARAMETER_ARG_CAN_NOT_ARG_PLEASE_SET_ARG_BECAUSE_ARG_749738D1 =
+ "Parameter %s can not be %s, please set to: %s. Because %s";
+ public static final String EXCEPTION_QUERY_EXECUTION_TIME_OUT_A5DC7BFB =
+ "Query execution is time out";
+ public static final String EXCEPTION_OBJECT_FILE_ARG_DOES_NOT_EXIST_7EA8CB1C
=
+ "Object file %s does not exist";
+ public static final String EXCEPTION_ARG_NOT_LEGAL_PRIVILEGE_504838E8 =
+ "%s is not a legal privilege";
public static final String EXCEPTION_SOME_PORTS_OCCUPIED_77ED044D = "Some
ports are occupied";
public static final String EXCEPTION_PORTS_ARG_OCCUPIED_B462E9DA = "Ports %s
are occupied";
- public static final String
EXCEPTION_UNEXPECTED_ERROR_OCCURS_SERIALIZATION_A6B2E222 = "Unexpected error
occurs in serialization";
- public static final String
EXCEPTION_COLUMN_ARG_TABLE_ARG_ARG_DOES_NOT_EXIST_D8145581 = "Column %s in
table '%s.%s' does not exist.";
- public static final String EXCEPTION_TABLE_ARG_ARG_DOES_NOT_EXIST_796E503B =
"Table '%s.%s' does not exist.";
- public static final String EXCEPTION_TABLE_ARG_ARG_ALREADY_EXISTS_D4BDF4B5 =
"Table '%s.%s' already exists.";
- public static final String
EXCEPTION_COULDN_T_CONSTRUCTOR_SERIESPARTITIONEXECUTOR_CLASS_ARG_34FB9F45 =
"Couldn't Constructor SeriesPartitionExecutor class: %s";
- public static final String
EXCEPTION_CANNOT_USE_SETVALUE_OBJECT_BEING_SET_ALREADY_MAP_676ED3BF = "Cannot
use setValue() when the object being set is already in the map";
- public static final String
EXCEPTION_ITERATOR_GETKEY_CAN_ONLY_CALLED_AFTER_NEXT_BEFORE_REMOVE_009C456B =
"Iterator getKey() can only be called after next() and before remove()";
- public static final String
EXCEPTION_ITERATOR_GETVALUE_CAN_ONLY_CALLED_AFTER_NEXT_BEFORE_REMOVE_927A88A2 =
"Iterator getValue() can only be called after next() and before remove()";
- public static final String
EXCEPTION_ITERATOR_SETVALUE_CAN_ONLY_CALLED_AFTER_NEXT_BEFORE_REMOVE_51505AD1 =
"Iterator setValue() can only be called after next() and before remove()";
- public static final String
LOG_FAILED_CLOSE_UDFCLASSLOADER_QUERYID_ARG_BECAUSE_ARG_8B1C3739 = "Failed to
close UDFClassLoader (queryId: {}), because {}";
- public static final String
EXCEPTION_ATTRIBUTE_ARG_ARG_REQUIRED_BUT_WAS_NOT_PROVIDED_CD090883 = "attribute
\"%s\"/\"%s\" is required but was not provided.";
- public static final String
EXCEPTION_USE_ATTRIBUTE_ARG_ARG_ONLY_ONE_AT_TIME_B431468C = "use attribute
\"%s\" or \"%s\" only one at a time.";
- public static final String
EXCEPTION_ILLEGAL_OUTLIER_METHOD_OUTLIER_TYPE_SHOULD_AVG_STENDIS_COS_PRENEXTDIS_91D1C70A
= "Illegal outlier method. Outlier type should be avg, stendis, cos or
prenextdis.";
- public static final String
EXCEPTION_ILLEGAL_AGGREGATION_METHOD_AGGREGATION_TYPE_SHOULD_AVG_MIN_MAX_SUM_2D7BEC96
= "Illegal aggregation method. Aggregation type should be avg, min, max, sum,
extreme, variance.";
- public static final String
EXCEPTION_CUMULATIVE_TABLE_FUNCTION_REQUIRES_SIZE_MUST_INTEGRAL_MULTIPLE_STEP_D8A9DA94
= "Cumulative table function requires size must be an integral multiple of
step.";
- public static final String
EXCEPTION_COLUMN_TYPE_MUST_NUMERIC_IF_DELTA_NOT_0_F7864D4E = " The column type
must be numeric if DELTA is not 0.";
- public static final String
EXCEPTION_TYPE_COLUMN_ARG_NOT_AS_EXPECTED_7A81636E = "The type of the column
[%s] is not as expected.";
- public static final String
EXCEPTION_REQUIRED_COLUMN_ARG_NOT_FOUND_SOURCE_TABLE_ARGUMENT_993E1C08 =
"Required column [%s] not found in the source table argument.";
- public static final String
EXCEPTION_UNSUPPORTED_PROGRESS_INDEX_TYPE_ARG_A84CDFF9 = "Unsupported progress
index type %s.";
- public static final String
EXCEPTION_TIMEWINDOWSTATEPROGRESSINDEX_DOES_NOT_SUPPORT_TOPOLOGICAL_SORTING_897C8976
= "TimeWindowStateProgressIndex does not support topological sorting";
- public static final String
EXCEPTION_INTENDED_READ_LENGTH_ARG_BUT_ARG_ACTUALLY_READ_DESERIALIZING_TIMEPROGRESSINDEX_63CD54E4
=
- "The intended read length is %s but %s is actually read when
deserializing TimeProgressIndex,"
- + " ProgressIndex: %s";
+ public static final String
EXCEPTION_UNEXPECTED_ERROR_OCCURS_SERIALIZATION_A6B2E222 =
+ "Unexpected error occurs in serialization";
+ public static final String
EXCEPTION_COLUMN_ARG_TABLE_ARG_ARG_DOES_NOT_EXIST_D8145581 =
+ "Column %s in table '%s.%s' does not exist.";
+ public static final String EXCEPTION_TABLE_ARG_ARG_DOES_NOT_EXIST_796E503B =
+ "Table '%s.%s' does not exist.";
+ public static final String EXCEPTION_TABLE_ARG_ARG_ALREADY_EXISTS_D4BDF4B5 =
+ "Table '%s.%s' already exists.";
+ public static final String
+
EXCEPTION_COULDN_T_CONSTRUCTOR_SERIESPARTITIONEXECUTOR_CLASS_ARG_34FB9F45 =
+ "Couldn't Constructor SeriesPartitionExecutor class: %s";
+ public static final String
EXCEPTION_CANNOT_USE_SETVALUE_OBJECT_BEING_SET_ALREADY_MAP_676ED3BF =
+ "Cannot use setValue() when the object being set is already in the map";
+ public static final String
+
EXCEPTION_ITERATOR_GETKEY_CAN_ONLY_CALLED_AFTER_NEXT_BEFORE_REMOVE_009C456B =
+ "Iterator getKey() can only be called after next() and before
remove()";
+ public static final String
+
EXCEPTION_ITERATOR_GETVALUE_CAN_ONLY_CALLED_AFTER_NEXT_BEFORE_REMOVE_927A88A2 =
+ "Iterator getValue() can only be called after next() and before
remove()";
+ public static final String
+
EXCEPTION_ITERATOR_SETVALUE_CAN_ONLY_CALLED_AFTER_NEXT_BEFORE_REMOVE_51505AD1 =
+ "Iterator setValue() can only be called after next() and before
remove()";
+ public static final String
LOG_FAILED_CLOSE_UDFCLASSLOADER_QUERYID_ARG_BECAUSE_ARG_8B1C3739 =
+ "Failed to close UDFClassLoader (queryId: {}), because {}";
+ public static final String
EXCEPTION_ATTRIBUTE_ARG_ARG_REQUIRED_BUT_WAS_NOT_PROVIDED_CD090883 =
+ "attribute \"%s\"/\"%s\" is required but was not provided.";
+ public static final String
EXCEPTION_USE_ATTRIBUTE_ARG_ARG_ONLY_ONE_AT_TIME_B431468C =
+ "use attribute \"%s\" or \"%s\" only one at a time.";
+ public static final String
+
EXCEPTION_ILLEGAL_OUTLIER_METHOD_OUTLIER_TYPE_SHOULD_AVG_STENDIS_COS_PRENEXTDIS_91D1C70A
=
+ "Illegal outlier method. Outlier type should be avg, stendis, cos or
prenextdis.";
+ public static final String
+
EXCEPTION_ILLEGAL_AGGREGATION_METHOD_AGGREGATION_TYPE_SHOULD_AVG_MIN_MAX_SUM_2D7BEC96
=
+ "Illegal aggregation method. Aggregation type should be avg, min,
max, sum, extreme, variance.";
+ public static final String
+
EXCEPTION_CUMULATIVE_TABLE_FUNCTION_REQUIRES_SIZE_MUST_INTEGRAL_MULTIPLE_STEP_D8A9DA94
=
+ "Cumulative table function requires size must be an integral
multiple of step.";
+ public static final String
EXCEPTION_COLUMN_TYPE_MUST_NUMERIC_IF_DELTA_NOT_0_F7864D4E =
+ " The column type must be numeric if DELTA is not 0.";
+ public static final String
EXCEPTION_TYPE_COLUMN_ARG_NOT_AS_EXPECTED_7A81636E =
+ "The type of the column [%s] is not as expected.";
+ public static final String
+ EXCEPTION_REQUIRED_COLUMN_ARG_NOT_FOUND_SOURCE_TABLE_ARGUMENT_993E1C08 =
+ "Required column [%s] not found in the source table argument.";
+ public static final String
EXCEPTION_UNSUPPORTED_PROGRESS_INDEX_TYPE_ARG_A84CDFF9 =
+ "Unsupported progress index type %s.";
+ public static final String
+
EXCEPTION_TIMEWINDOWSTATEPROGRESSINDEX_DOES_NOT_SUPPORT_TOPOLOGICAL_SORTING_897C8976
=
+ "TimeWindowStateProgressIndex does not support topological sorting";
+ public static final String
+
EXCEPTION_INTENDED_READ_LENGTH_ARG_BUT_ARG_ACTUALLY_READ_DESERIALIZING_TIMEPROGRESSINDEX_63CD54E4
=
+ "The intended read length is %s but %s is actually read when
deserializing TimeProgressIndex,"
+ + " ProgressIndex: %s";
public static final String EXCEPTION_COLON_3A291246 = " : ";
- public static final String EXCEPTION_DATAPARTITIONMAP_IS_NULL_B764418A =
"dataPartitionMap is null";
+ public static final String EXCEPTION_DATAPARTITIONMAP_IS_NULL_B764418A =
+ "dataPartitionMap is null";
public static final String EXCEPTION_ARG_634FCEDB = "%s";
- public static final String
EXCEPTION_TABLE_ARGUMENT_WITH_SET_SEMANTICS_REQUIRES_AN_ORDER_BY_CLAUSE_10C986D9
= "Table argument with set semantics requires an ORDER BY clause.";
- public static final String
EXCEPTION_THE_TYPE_OF_THE_COLUMN_ARG_IS_NOT_COMPARABLE_E3098096 = "The type of
the column [%s] is not comparable.";
- public static final String
EXCEPTION_NO_COMPARABLE_COLUMNS_FOUND_FOR_M4_CALCULATION_4E5A3092 = "No
comparable columns found for M4 calculation.";
- public static final String
EXCEPTION_INVALID_SCALAR_ARGUMENT_SLIDE_SHOULD_BE_A_POSITIVE_VALUE_F019E091 =
"Invalid scalar argument SLIDE, should be a positive value";
- public static final String
EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_THE_TIME_COLUMN_SPECIFIED_BY_THE_TIMECOL_ARGUMENT_4375BAE9
= "The ORDER BY clause of the DATA argument must contain exactly the time
column specified by the TIMECOL argument.";
- public static final String EXCEPTION_UNSUPPORTED_M4_VALUE_TYPE_AF0EF286 =
"Unsupported M4 value type: ";
- public static final String
EXCEPTION_DISK_SPACE_WARNING_THRESHOLD_MUST_BE_IN_0_1_BUT_WAS_7B345766 =
"disk_space_warning_threshold must be in [0, 1), but was ";
+ public static final String
+
EXCEPTION_TABLE_ARGUMENT_WITH_SET_SEMANTICS_REQUIRES_AN_ORDER_BY_CLAUSE_10C986D9
=
+ "Table argument with set semantics requires an ORDER BY clause.";
+ public static final String
EXCEPTION_THE_TYPE_OF_THE_COLUMN_ARG_IS_NOT_COMPARABLE_E3098096 =
+ "The type of the column [%s] is not comparable.";
+ public static final String
EXCEPTION_NO_COMPARABLE_COLUMNS_FOUND_FOR_M4_CALCULATION_4E5A3092 =
+ "No comparable columns found for M4 calculation.";
+ public static final String
+
EXCEPTION_INVALID_SCALAR_ARGUMENT_SLIDE_SHOULD_BE_A_POSITIVE_VALUE_F019E091 =
+ "Invalid scalar argument SLIDE, should be a positive value";
+ public static final String
+
EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_THE_TIME_COLUMN_SPECIFIED_BY_THE_TIMECOL_ARGUMENT_4375BAE9
=
+ "The ORDER BY clause of the DATA argument must contain exactly the
time column specified by the TIMECOL argument.";
+ public static final String EXCEPTION_UNSUPPORTED_M4_VALUE_TYPE_AF0EF286 =
+ "Unsupported M4 value type: ";
+ public static final String
+ EXCEPTION_DISK_SPACE_WARNING_THRESHOLD_MUST_BE_IN_0_1_BUT_WAS_7B345766 =
+ "disk_space_warning_threshold must be in [0, 1), but was ";
+ public static final String EXCEPTION_FILTER_FUNCTION_WPASS_VALIDATION =
+ "the value of wpass should be in (0, 1)";
+ public static final String EXCEPTION_NO_CALCULATE_COLUMNS = "No columns
could be calculated.";
+ public static final String EXCEPTION_NOT_ALLOWED_COLUMNS =
+ "Only column with double, float, int32, int64 can be calculated by the
function, %s is the %s.";
public static final String
LOG_TRUSTED_CHANNEL_FUNCTION_FAILED_INITIATOR_ARG_TARGET_ARG_E4C28443 =
"Trusted channel function failed: initiator=%s, target=%s";
-
+ public static final String
EXCEPTION_FILTER_FUNCTION_ROW_INDEX_EXCEED_MAXIMUM =
+ "row index exceeds the maximum allowed number in one partition";
+ public static final String
+
EXCEPTION_XCORR_REQUIRES_EXACTLY_TWO_CALCULATION_COLUMNS_BUT_FOUND_ARG_2FF8EB0C
=
+ "XCorr requires exactly two calculation columns, but found %d.";
+ public static final String EXCEPTION_COLUMN_LACK_OF_NAME = "the column in
table lack of the name";
}
diff --git
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
index 07d886a44f3..6cddaa0b1fc 100644
---
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
@@ -217,5 +217,12 @@ public final class CommonMessages {
public static final String
EXCEPTION_DISK_SPACE_WARNING_THRESHOLD_MUST_BE_IN_0_1_BUT_WAS_7B345766 =
"disk_space_warning_threshold 必须在 [0, 1) 范围内，但实际为 ";
public static final String
LOG_TRUSTED_CHANNEL_FUNCTION_FAILED_INITIATOR_ARG_TARGET_ARG_E4C28443 =
"可信信道功能失效：发起者=%s，目标端=%s";
-
+ public static final String EXCEPTION_FILTER_FUNCTION_WPASS_VALIDATION =
"wpass的取值范围应该位于(0, 1)";
+ public static final String EXCEPTION_NO_CALCULATE_COLUMNS = "没有找到可以计算的列.";
+ public static final String EXCEPTION_NOT_ALLOWED_COLUMNS = "只允许列类型为double,
float, int32, int64参与函数计算, 当前列 %s 类型是 %s.";
+ public static final String
EXCEPTION_FILTER_FUNCTION_ROW_INDEX_EXCEED_MAXIMUM = "分区行数超过了最大限制";
+ public static final String
+
EXCEPTION_XCORR_REQUIRES_EXACTLY_TWO_CALCULATION_COLUMNS_BUT_FOUND_ARG_2FF8EB0C
=
+ "XCorr 要求必须正好有两列计算列，但实际找到 %d 列。";
+ public static final String EXCEPTION_COLUMN_LACK_OF_NAME = "表参数中列缺少名字";
}
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 decaf8dc466..818747bb714 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
@@ -27,10 +27,13 @@ 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.HighPassTableFunction;
+import
org.apache.iotdb.commons.udf.builtin.relational.tvf.LowPassTableFunction;
import org.apache.iotdb.commons.udf.builtin.relational.tvf.M4TableFunction;
import
org.apache.iotdb.commons.udf.builtin.relational.tvf.SessionTableFunction;
import org.apache.iotdb.commons.udf.builtin.relational.tvf.TumbleTableFunction;
import
org.apache.iotdb.commons.udf.builtin.relational.tvf.VariationTableFunction;
+import org.apache.iotdb.commons.udf.builtin.relational.tvf.XCorrTableFunction;
import org.apache.iotdb.udf.api.relational.TableFunction;
import java.util.Arrays;
@@ -49,7 +52,10 @@ public enum TableBuiltinTableFunction {
FFT("fft"),
FORECAST("forecast"),
PATTERN_MATCH("pattern_match"),
- CLASSIFY("classify");
+ CLASSIFY("classify"),
+ LOWPASS("lowpass"),
+ HIGHPASS("highpass"),
+ XCORR("xcorr");
private final String functionName;
@@ -99,6 +105,12 @@ public enum TableBuiltinTableFunction {
return new ForecastTableFunction();
case "classify":
return new ClassifyTableFunction();
+ case "lowpass":
+ return new LowPassTableFunction();
+ case "highpass":
+ return new HighPassTableFunction();
+ case "xcorr":
+ return new XCorrTableFunction();
default:
throw new UnsupportedOperationException(
String.format(QueryMessages.UNSUPPORTED_TABLE_FUNCTION,
functionName));
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FilterTransferTableFunction.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FilterTransferTableFunction.java
new file mode 100644
index 00000000000..0fe09c60cb7
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FilterTransferTableFunction.java
@@ -0,0 +1,315 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.udf.builtin.relational.tvf;
+
+import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.i18n.CommonMessages;
+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 java.util.ArrayList;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static
org.apache.iotdb.commons.i18n.CommonMessages.EXCEPTION_FILTER_FUNCTION_WPASS_VALIDATION;
+
+public abstract class FilterTransferTableFunction implements TableFunction {
+
+ public static final String DATA_PARAMETER_NAME = "DATA";
+ public static final String TIMECOL_PARAMETER_NAME = "TIMECOL";
+ public static final String WPASS = "WPASS";
+
+ protected static final String PARTITION_TYPES_PROPERTY = "PARTITION_TYPES";
+ protected static final String CALCULATION_COLUMN_COUNT_PROPERTY =
"CALCULATION_COLUMN_COUNT";
+
+ @Override
+ public List<ParameterSpecification> getArgumentsSpecifications() {
+ return Arrays.asList(
+
TableParameterSpecification.builder().name(DATA_PARAMETER_NAME).setSemantics().build(),
+ ScalarParameterSpecification.builder()
+ .name(TIMECOL_PARAMETER_NAME)
+ .type(Type.STRING)
+ .build(),
+ ScalarParameterSpecification.builder()
+ .name(WPASS)
+ .type(Type.DOUBLE)
+ .addChecker(
+ object -> {
+ if (object instanceof Number) {
+ double value = ((Number) object).doubleValue();
+ if (value > 0 && value < 1) {
+ return null;
+ }
+ }
+ return EXCEPTION_FILTER_FUNCTION_WPASS_VALIDATION;
+ })
+ .build());
+ }
+
+ @Override
+ public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws
UDFException {
+
+ // order by column must only be the time column
+ int timeColumnIndex =
+ WindowTVFUtils.checkOrderByColumn(arguments, DATA_PARAMETER_NAME,
TIMECOL_PARAMETER_NAME);
+ TableArgument tableArgument = (TableArgument)
arguments.get(DATA_PARAMETER_NAME);
+
+ List<Integer> partitionIndexes =
WindowTVFUtils.getPartitionIndexes(tableArgument);
+ Set<Integer> excludedIndexes = new HashSet<>(partitionIndexes);
+ excludedIndexes.add(timeColumnIndex);
+
+ List<Type> partitionTypes = new ArrayList<>();
+ List<Integer> calculationIndexes = new ArrayList<>();
+ DescribedSchema.Builder schemaBuilder = new DescribedSchema.Builder();
+
+ // record the partition columns
+ for (int partitionIndex : partitionIndexes) {
+ Type type = tableArgument.getFieldTypes().get(partitionIndex);
+ partitionTypes.add(type);
+
schemaBuilder.addField(tableArgument.getFieldNames().get(partitionIndex).get(),
type);
+ }
+
+ // record the time column
+ schemaBuilder.addField(tableArgument.getFieldNames().get(timeColumnIndex),
Type.TIMESTAMP);
+ // record the calculation columns, only double, float, int32, and int64
are allowed
+ calculationIndexes.addAll(
+ WindowTVFUtils.getCalculationIndexes(
+ tableArgument,
+ excludedIndexes,
+ columnName ->
schemaBuilder.addField(convertColumnName(columnName), Type.DOUBLE)));
+
+ if (calculationIndexes.isEmpty()) {
+ throw new
SemanticException(CommonMessages.EXCEPTION_NO_CALCULATE_COLUMNS);
+ }
+
+ MapTableFunctionHandle.Builder handleBuilder =
+ new MapTableFunctionHandle.Builder()
+ .addProperty(PARTITION_TYPES_PROPERTY,
WindowTVFUtils.joinTypes(partitionTypes))
+ .addProperty(CALCULATION_COLUMN_COUNT_PROPERTY,
calculationIndexes.size())
+ .addProperty(WPASS, ((ScalarArgument)
arguments.get(WPASS)).getValue());
+ List<Integer> requiredColumns = new ArrayList<>(partitionIndexes);
+ requiredColumns.add(timeColumnIndex);
+ requiredColumns.addAll(calculationIndexes);
+
+ return TableFunctionAnalysis.builder()
+ .properColumnSchema(schemaBuilder.build())
+ .requireRecordSnapshot(false)
+ .requiredColumns(DATA_PARAMETER_NAME, requiredColumns)
+ .handle(handleBuilder.build())
+ .build();
+ }
+
+ @Override
+ public TableFunctionHandle createTableFunctionHandle() {
+ return new MapTableFunctionHandle();
+ }
+
+ @Override
+ public abstract TableFunctionProcessorProvider getProcessorProvider(
+ TableFunctionHandle tableFunctionHandle);
+
+ protected abstract String convertColumnName(String columnName);
+
+ /**
+ * Processes one complete partition (which may contain multiple calculation
columns).
+ *
+ * <p>FFT-based filters cannot operate on null values. We therefore collect
only finite, non-null
+ * values for each calculation column and keep their original row positions.
The compact sequence
+ * is passed to the filter, then the transformed values are expanded back to
the partition's
+ * original row layout. Rows that had a null or non-finite input remain null
in the output; they
+ * do not participate in the FFT calculation.
+ */
+ protected abstract static class FilterTransferDataProcessor
+ implements TableFunctionDataProcessor {
+
+ protected static final int INITIAL_CAPACITY = 512;
+ protected static final int MAX_COUNT_IN_ONE_PARTITION = 65536;
+
+ private final double wpass;
+ private int partitionRowCount;
+
+ private final int partitionColumnCount;
+ private final int timeColumnIndex;
+ private final int calculationColumnStartIndex;
+ private final Type[] partitionTypes;
+ private final Object[] partitionValues;
+
+ // CalculationColumnContainer collect the all value of a column in one
partition
+ private long[] partitionTimestamps;
+ private final CalculationColumnContainer[] calculationColumnContainers;
+
+ protected FilterTransferDataProcessor(
+ double wpass, Type[] partitionTypes, int calculationColumnCount) {
+ this.wpass = wpass;
+ this.partitionColumnCount = partitionTypes.length;
+ this.timeColumnIndex = partitionColumnCount;
+ this.partitionRowCount = 0;
+ this.calculationColumnStartIndex = timeColumnIndex + 1;
+ this.partitionTypes = partitionTypes;
+ this.partitionValues = new Object[partitionTypes.length];
+ this.partitionTimestamps = new long[INITIAL_CAPACITY];
+ this.calculationColumnContainers = new
CalculationColumnContainer[calculationColumnCount];
+ for (int i = 0; i < calculationColumnCount; i++) {
+ calculationColumnContainers[i] = new CalculationColumnContainer();
+ }
+ }
+
+ @Override
+ public void process(
+ Record input,
+ List<ColumnBuilder> properColumnBuilders,
+ ColumnBuilder passThroughIndexBuilder) {
+ if (partitionRowCount >= MAX_COUNT_IN_ONE_PARTITION) {
+ throw new SemanticException(
+ CommonMessages.EXCEPTION_FILTER_FUNCTION_ROW_INDEX_EXCEED_MAXIMUM);
+ }
+ if (partitionRowCount == 0) {
+ capturePartitionValues(input);
+ }
+ collectTimeColumnValue(input);
+ collectCalculationValues(input, partitionRowCount);
+ partitionRowCount++;
+ }
+
+ private void capturePartitionValues(Record input) {
+ for (int i = 0; i < partitionColumnCount; i++) {
+ partitionValues[i] =
+ input.isNull(i) ? null : WindowTVFUtils.readValue(input, i,
partitionTypes[i]);
+ }
+ }
+
+ private void collectTimeColumnValue(Record input) {
+ if (partitionRowCount >= partitionTimestamps.length) {
+ int newCapacity = partitionTimestamps.length +
(partitionTimestamps.length >> 2);
+ partitionTimestamps = Arrays.copyOf(partitionTimestamps, newCapacity);
+ }
+ partitionTimestamps[partitionRowCount] = input.getLong(timeColumnIndex);
+ }
+
+ private void collectCalculationValues(Record input, int partitionRowIndex)
{
+ for (int i = 0; i < calculationColumnContainers.length; i++) {
+ // Missing and non-finite values are intentionally excluded from the
compact FFT input.
+ if (!input.isNull(calculationColumnStartIndex + i)) {
+ double aDouble = input.getDouble(calculationColumnStartIndex + i);
+ if (Double.isFinite(aDouble)) {
+ calculationColumnContainers[i].add(partitionRowIndex, aDouble);
+ }
+ ;
+ }
+ }
+ }
+
+ @Override
+ public void finish(
+ List<ColumnBuilder> properColumnBuilders, ColumnBuilder
passThroughIndexBuilder) {
+
+ // collect the partition columns
+ for (int columnIndex = 0; columnIndex < partitionColumnCount;
columnIndex++) {
+ ColumnBuilder partitionColumnBuilder =
properColumnBuilders.get(columnIndex);
+ Object partitionValue = partitionValues[columnIndex];
+ Type partitionType = partitionTypes[columnIndex];
+ for (int rowIndex = 0; rowIndex < partitionRowCount; rowIndex++) {
+ WindowTVFUtils.writeValue(partitionColumnBuilder, partitionValue,
partitionType);
+ }
+ }
+
+ // collect the time column
+ ColumnBuilder timeColumnBuilder =
properColumnBuilders.get(timeColumnIndex);
+ for (int rowIndex = 0; rowIndex < partitionRowCount; rowIndex++) {
+ timeColumnBuilder.writeLong(partitionTimestamps[rowIndex]);
+ }
+
+ // collect the calculation column
+ for (int i = 0; i < calculationColumnContainers.length; i++) {
+ transformSingleColumn(
+ calculationColumnContainers[i],
+ wpass,
+ properColumnBuilders.get(calculationColumnStartIndex + i));
+ }
+ }
+
+ private void transformSingleColumn(
+ CalculationColumnContainer columnContainer,
+ double wpass,
+ ColumnBuilder properColumnBuilder) {
+ int size = columnContainer.validValueCount;
+ if (size == 0) {
+ for (int rowIndex = 0; rowIndex < partitionRowCount; rowIndex++) {
+ properColumnBuilder.appendNull();
+ }
+ return;
+ }
+ double[] temp = filterTransform(columnContainer, size, wpass);
+ // Restore the transformed values to their original rows; excluded rows
stay null.
+ int validValueIndex = 0;
+ for (int i = 0; i < partitionRowCount; i++) {
+ if (columnContainer.validRows.get(i)) {
+ properColumnBuilder.writeDouble(temp[2 * validValueIndex]);
+ validValueIndex++;
+ } else {
+ properColumnBuilder.appendNull();
+ }
+ }
+ }
+
+ protected abstract double[] filterTransform(
+ CalculationColumnContainer columnContainer, int size, double wpass);
+ }
+
+ protected static class CalculationColumnContainer {
+ protected double[] validValues = new
double[FilterTransferDataProcessor.INITIAL_CAPACITY];
+ private int validValueCount = 0;
+ private final BitSet validRows = new BitSet();
+
+ public void add(int rowIndex, double value) {
+ ensureCapacity(validValueCount + 1);
+ validValues[validValueCount++] = value;
+ validRows.set(rowIndex);
+ }
+
+ private void ensureCapacity(int requiredCapacity) {
+ if (requiredCapacity <= validValues.length) {
+ return;
+ }
+ int newCapacity = validValues.length + (validValues.length >> 1);
+ validValues = Arrays.copyOf(validValues, newCapacity);
+ }
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/HighPassTableFunction.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/HighPassTableFunction.java
new file mode 100644
index 00000000000..88fa37bee62
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/HighPassTableFunction.java
@@ -0,0 +1,81 @@
+/*
+ * 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.udf.builtin.relational.tvf.fft.DoubleFFT_1D;
+import org.apache.iotdb.udf.api.relational.table.MapTableFunctionHandle;
+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.processor.TableFunctionDataProcessor;
+import org.apache.iotdb.udf.api.type.Type;
+
+public class HighPassTableFunction extends FilterTransferTableFunction {
+
+ @Override
+ public TableFunctionProcessorProvider getProcessorProvider(
+ TableFunctionHandle tableFunctionHandle) {
+ MapTableFunctionHandle handle = (MapTableFunctionHandle)
tableFunctionHandle;
+ double wpass = (double) handle.getProperty(WPASS);
+ Type[] partitionTypes =
+ WindowTVFUtils.parseTypes((String)
handle.getProperty(PARTITION_TYPES_PROPERTY));
+ int calculationColumnCount = (Integer)
handle.getProperty(CALCULATION_COLUMN_COUNT_PROPERTY);
+
+ return new TableFunctionProcessorProvider() {
+ @Override
+ public TableFunctionDataProcessor getDataProcessor() {
+ return new HighPassDataProcessor(wpass, partitionTypes,
calculationColumnCount);
+ }
+ };
+ }
+
+ @Override
+ protected String convertColumnName(String columnName) {
+ return String.format("highpass(%s)", columnName);
+ }
+
+ protected static class HighPassDataProcessor extends
FilterTransferDataProcessor {
+
+ public HighPassDataProcessor(double wpass, Type[] partitionTypes, int
calculationColumnCount) {
+ super(wpass, partitionTypes, calculationColumnCount);
+ }
+
+ @Override
+ protected double[] filterTransform(
+ CalculationColumnContainer columnContainer, int size, double wpass) {
+ DoubleFFT_1D fft = new DoubleFFT_1D(size);
+ double[] temp = new double[2 * size];
+ for (int i = 0; i < size; i++) {
+ temp[2 * i] = columnContainer.validValues[i];
+ temp[2 * i + 1] = 0;
+ }
+
+ fft.complexForward(temp);
+ int m = (int) Math.floor(wpass * size / 2);
+ for (int i = 0; i <= 2 * m + 1; i++) {
+ temp[i] = 0;
+ }
+ for (int i = 2 * (size - m); i < 2 * size; i++) {
+ temp[i] = 0;
+ }
+ fft.complexInverse(temp, true);
+ return temp;
+ }
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LowPassTableFunction.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LowPassTableFunction.java
new file mode 100644
index 00000000000..b563af717a4
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LowPassTableFunction.java
@@ -0,0 +1,77 @@
+/*
+ * 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.udf.builtin.relational.tvf.fft.DoubleFFT_1D;
+import org.apache.iotdb.udf.api.relational.table.MapTableFunctionHandle;
+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.processor.TableFunctionDataProcessor;
+import org.apache.iotdb.udf.api.type.Type;
+
+public class LowPassTableFunction extends FilterTransferTableFunction {
+
+ @Override
+ public TableFunctionProcessorProvider getProcessorProvider(
+ TableFunctionHandle tableFunctionHandle) {
+ MapTableFunctionHandle handle = (MapTableFunctionHandle)
tableFunctionHandle;
+ double wpass = (double) handle.getProperty(WPASS);
+ Type[] partitionTypes =
+ WindowTVFUtils.parseTypes((String)
handle.getProperty(PARTITION_TYPES_PROPERTY));
+ int calculationColumnCount = (Integer)
handle.getProperty(CALCULATION_COLUMN_COUNT_PROPERTY);
+
+ return new TableFunctionProcessorProvider() {
+ @Override
+ public TableFunctionDataProcessor getDataProcessor() {
+ return new LowPassDataProcessor(wpass, partitionTypes,
calculationColumnCount);
+ }
+ };
+ }
+
+ @Override
+ protected String convertColumnName(String columnName) {
+ return String.format("lowpass(%s)", columnName);
+ }
+
+ protected static class LowPassDataProcessor extends
FilterTransferDataProcessor {
+
+ public LowPassDataProcessor(double wpass, Type[] partitionTypes, int
calculationColumnCount) {
+ super(wpass, partitionTypes, calculationColumnCount);
+ }
+
+ @Override
+ protected double[] filterTransform(
+ CalculationColumnContainer columnContainer, int size, double wpass) {
+ DoubleFFT_1D fft = new DoubleFFT_1D(size);
+ double[] temp = new double[2 * size];
+ for (int i = 0; i < size; i++) {
+ temp[2 * i] = columnContainer.validValues[i];
+ temp[2 * i + 1] = 0;
+ }
+ fft.complexForward(temp);
+ int m = (int) Math.ceil(wpass * size / 2);
+ for (int i = 2 * m; i <= 2 * (size - m) + 1; i++) {
+ temp[i] = 0;
+ }
+ fft.complexInverse(temp, true);
+ return temp;
+ }
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/M4TableFunction.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/M4TableFunction.java
index 733696b6212..7b508fcb53d 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/M4TableFunction.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/M4TableFunction.java
@@ -53,7 +53,6 @@ import java.util.List;
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 M4TableFunction implements TableFunction {
@@ -72,19 +71,6 @@ public class M4TableFunction implements TableFunction {
private static final String PARTICIPANT_TYPES_PROPERTY =
"__M4_PARTICIPANT_TYPES";
private static final long UNSPECIFIED_SLIDE = Long.MIN_VALUE;
private static final long INVALID_INDEX = -1;
- private static final Set<Type> SUPPORTED_PARTITION_TYPES =
- new HashSet<>(
- Arrays.asList(
- Type.BOOLEAN,
- Type.INT32,
- Type.INT64,
- Type.FLOAT,
- Type.DOUBLE,
- Type.TEXT,
- Type.TIMESTAMP,
- Type.DATE,
- Type.BLOB,
- Type.STRING));
@Override
public List<ParameterSpecification> getArgumentsSpecifications() {
@@ -113,31 +99,20 @@ public class M4TableFunction implements TableFunction {
@Override
public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws
UDFException {
- TableArgument tableArgument = (TableArgument)
arguments.get(DATA_PARAMETER_NAME);
- if (tableArgument.getOrderBy().isEmpty()) {
- throw new SemanticException(
- CommonMessages
-
.EXCEPTION_TABLE_ARGUMENT_WITH_SET_SEMANTICS_REQUIRES_AN_ORDER_BY_CLAUSE_10C986D9);
- }
-
- String timeColumn =
- (String) ((ScalarArgument)
arguments.get(TIMECOL_PARAMETER_NAME)).getValue();
int timeColumnIndex =
- findColumnIndex(tableArgument, timeColumn,
Collections.singleton(Type.TIMESTAMP));
- validateOrderBy(tableArgument, timeColumn);
-
- List<Integer> partitionIndexes = getPartitionIndexes(tableArgument);
+ WindowTVFUtils.checkOrderByColumn(arguments, DATA_PARAMETER_NAME,
TIMECOL_PARAMETER_NAME);
+ TableArgument tableArgument = (TableArgument)
arguments.get(DATA_PARAMETER_NAME);
+ List<Integer> partitionIndexes =
WindowTVFUtils.getPartitionIndexes(tableArgument);
Set<Integer> excludedIndexes = new HashSet<>(partitionIndexes);
excludedIndexes.add(timeColumnIndex);
- boolean isTimeWindow =
- arguments.containsKey(WINDOW_MODE_PARAMETER_NAME)
- && (boolean) ((ScalarArgument)
arguments.get(WINDOW_MODE_PARAMETER_NAME)).getValue();
-
List<Integer> participantIndexes = new ArrayList<>();
List<Type> partitionTypes = new ArrayList<>();
List<Type> participantTypes = new ArrayList<>();
DescribedSchema.Builder schemaBuilder = new DescribedSchema.Builder();
+ boolean isTimeWindow =
+ arguments.containsKey(WINDOW_MODE_PARAMETER_NAME)
+ && (boolean) ((ScalarArgument)
arguments.get(WINDOW_MODE_PARAMETER_NAME)).getValue();
if (isTimeWindow) {
schemaBuilder
.addField(OUTPUT_WINDOW_START_COLUMN, Type.TIMESTAMP)
@@ -189,8 +164,8 @@ public class M4TableFunction implements TableFunction {
.addProperty(WINDOW_MODE_PARAMETER_NAME, isTimeWindow)
.addProperty(SIZE_PARAMETER_NAME, size)
.addProperty(SLIDE_PARAMETER_NAME, slide)
- .addProperty(PARTITION_TYPES_PROPERTY, joinTypes(partitionTypes))
- .addProperty(PARTICIPANT_TYPES_PROPERTY,
joinTypes(participantTypes));
+ .addProperty(PARTITION_TYPES_PROPERTY,
WindowTVFUtils.joinTypes(partitionTypes))
+ .addProperty(PARTICIPANT_TYPES_PROPERTY,
WindowTVFUtils.joinTypes(participantTypes));
if (isTimeWindow) {
handleBuilder.addProperty(
ORIGIN_PARAMETER_NAME,
@@ -224,8 +199,10 @@ public class M4TableFunction implements TableFunction {
long size = (long) handle.getProperty(SIZE_PARAMETER_NAME);
long slide = (long) handle.getProperty(SLIDE_PARAMETER_NAME);
long origin = isTimeWindow ? (long)
handle.getProperty(ORIGIN_PARAMETER_NAME) : 0L;
- Type[] partitionTypes = parseTypes((String)
handle.getProperty(PARTITION_TYPES_PROPERTY));
- Type[] participantTypes = parseTypes((String)
handle.getProperty(PARTICIPANT_TYPES_PROPERTY));
+ Type[] partitionTypes =
+ WindowTVFUtils.parseTypes((String)
handle.getProperty(PARTITION_TYPES_PROPERTY));
+ Type[] participantTypes =
+ WindowTVFUtils.parseTypes((String)
handle.getProperty(PARTICIPANT_TYPES_PROPERTY));
return new TableFunctionProcessorProvider() {
@Override
@@ -240,51 +217,11 @@ public class M4TableFunction implements TableFunction {
};
}
- private static void validateOrderBy(TableArgument tableArgument, String
timeColumn) {
- if (tableArgument.getOrderBy().size() != 1
- || !tableArgument.getOrderBy().get(0).equalsIgnoreCase(timeColumn)) {
- throw new SemanticException(
- CommonMessages
-
.EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_THE_TIME_COLUMN_SPECIFIED_BY_THE_TIMECOL_ARGUMENT_4375BAE9);
- }
- }
-
- private static List<Integer> getPartitionIndexes(TableArgument
tableArgument) {
- List<Integer> indexes = new ArrayList<>();
- for (String partitionColumn : tableArgument.getPartitionBy()) {
- indexes.add(findColumnIndex(tableArgument, partitionColumn,
SUPPORTED_PARTITION_TYPES));
- }
- return indexes;
- }
-
// BLOB can be used as a partition column because M4 only needs to
read/write it there
private static boolean isComparableType(Type type) {
return type != Type.BLOB && type != Type.OBJECT;
}
- private static String joinTypes(List<Type> types) {
- StringBuilder builder = new StringBuilder();
- for (int i = 0; i < types.size(); i++) {
- if (i > 0) {
- builder.append(',');
- }
- builder.append(types.get(i).name());
- }
- return builder.toString();
- }
-
- private static Type[] parseTypes(String value) {
- if (value.isEmpty()) {
- return new Type[0];
- }
- String[] values = value.split(",");
- Type[] types = new Type[values.length];
- for (int i = 0; i < values.length; i++) {
- types[i] = Type.valueOf(values[i]);
- }
- return types;
- }
-
private static M4Column[] createColumns(Type[] types, int firstInputIndex) {
M4Column[] columns = new M4Column[types.length];
for (int i = 0; i < types.length; i++) {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/WindowTVFUtils.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/WindowTVFUtils.java
index 3271a30239a..36eae02cf64 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/WindowTVFUtils.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/WindowTVFUtils.java
@@ -19,17 +19,50 @@
package org.apache.iotdb.commons.udf.builtin.relational.tvf;
+import org.apache.iotdb.commons.exception.SemanticException;
import org.apache.iotdb.commons.i18n.CommonMessages;
import org.apache.iotdb.udf.api.exception.UDFColumnNotFoundException;
import org.apache.iotdb.udf.api.exception.UDFException;
import org.apache.iotdb.udf.api.exception.UDFTypeMismatchException;
+import org.apache.iotdb.udf.api.relational.access.Record;
+import org.apache.iotdb.udf.api.relational.table.argument.Argument;
+import org.apache.iotdb.udf.api.relational.table.argument.ScalarArgument;
import org.apache.iotdb.udf.api.relational.table.argument.TableArgument;
import org.apache.iotdb.udf.api.type.Type;
+import org.apache.tsfile.block.column.ColumnBuilder;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.function.Consumer;
+
+import static
org.apache.iotdb.commons.i18n.CommonMessages.EXCEPTION_COLUMN_LACK_OF_NAME;
public class WindowTVFUtils {
+
+ private static final Set<Type> ALLOWED_CALCULATION_TYPES =
+ Set.of(Type.DOUBLE, Type.FLOAT, Type.INT32, Type.INT64);
+
+ private static final Set<Type> SUPPORTED_PARTITION_TYPES =
+ new HashSet<>(
+ Arrays.asList(
+ Type.BOOLEAN,
+ Type.INT32,
+ Type.INT64,
+ Type.FLOAT,
+ Type.DOUBLE,
+ Type.TEXT,
+ Type.TIMESTAMP,
+ Type.DATE,
+ Type.BLOB,
+ Type.STRING));
+
/**
* Find the index of the column in the table argument.
*
@@ -58,4 +91,156 @@ public class WindowTVFUtils {
CommonMessages.EXCEPTION_REQUIRED_COLUMN_ARG_NOT_FOUND_SOURCE_TABLE_ARGUMENT_993E1C08,
expectedFieldName));
}
+
+ public static void validateOrderBy(TableArgument tableArgument, String
timeColumn) {
+ if (tableArgument.getOrderBy().size() != 1
+ || !tableArgument.getOrderBy().get(0).equalsIgnoreCase(timeColumn)) {
+ throw new SemanticException(
+ CommonMessages
+
.EXCEPTION_THE_ORDER_BY_CLAUSE_OF_THE_DATA_ARGUMENT_MUST_CONTAIN_EXACTLY_THE_TIME_COLUMN_SPECIFIED_BY_THE_TIMECOL_ARGUMENT_4375BAE9);
+ }
+ }
+
+ public static List<Integer> getPartitionIndexes(TableArgument tableArgument)
{
+ List<Integer> indexes = new ArrayList<>();
+ for (String partitionColumn : tableArgument.getPartitionBy()) {
+ indexes.add(findColumnIndex(tableArgument, partitionColumn,
SUPPORTED_PARTITION_TYPES));
+ }
+ return indexes;
+ }
+
+ /**
+ * Collect calculation-column indexes after excluding partition and time
columns.
+ *
+ * <p>If {@code calculationColumnConsumer} is provided, it is invoked with
each calculation column
+ * name so the caller can append the corresponding result field to its
output schema.
+ */
+ public static List<Integer> getCalculationIndexes(
+ TableArgument tableArgument,
+ Set<Integer> excludedIndexes,
+ Consumer<String> calculationColumnConsumer) {
+ List<Integer> calculationIndexes = new ArrayList<>();
+ for (int i = 0; i < tableArgument.getFieldTypes().size(); i++) {
+ if (excludedIndexes.contains(i)) {
+ continue;
+ }
+
+ Type type = tableArgument.getFieldTypes().get(i);
+ String columnName =
+ tableArgument
+ .getFieldNames()
+ .get(i)
+ .orElseThrow(() -> new
SemanticException(EXCEPTION_COLUMN_LACK_OF_NAME));
+ if (!ALLOWED_CALCULATION_TYPES.contains(type)) {
+ throw new SemanticException(
+ String.format(CommonMessages.EXCEPTION_NOT_ALLOWED_COLUMNS,
columnName, type));
+ }
+
+ calculationIndexes.add(i);
+ if (calculationColumnConsumer != null) {
+ calculationColumnConsumer.accept(columnName);
+ }
+ }
+ return calculationIndexes;
+ }
+
+ public static String joinTypes(List<Type> types) {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < types.size(); i++) {
+ if (i > 0) {
+ builder.append(',');
+ }
+ builder.append(types.get(i).name());
+ }
+ return builder.toString();
+ }
+
+ /** check the order by column is the timeColumn */
+ public static int checkOrderByColumn(
+ Map<String, Argument> arguments, String dataParameterName, String
timeParameterName) {
+ TableArgument tableArgument = (TableArgument)
arguments.get(dataParameterName);
+ if (tableArgument.getOrderBy().isEmpty()) {
+ throw new SemanticException(
+ CommonMessages
+
.EXCEPTION_TABLE_ARGUMENT_WITH_SET_SEMANTICS_REQUIRES_AN_ORDER_BY_CLAUSE_10C986D9);
+ }
+
+ String timeColumn = (String) ((ScalarArgument)
arguments.get(timeParameterName)).getValue();
+ int timeColumnIndex =
+ findColumnIndex(tableArgument, timeColumn,
Collections.singleton(Type.TIMESTAMP));
+ WindowTVFUtils.validateOrderBy(tableArgument, timeColumn);
+ return timeColumnIndex;
+ }
+
+ public 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;
+ }
+
+ public static Object readValue(Record input, int columnIndex, Type
partitionType) {
+ switch (partitionType) {
+ case BOOLEAN:
+ return input.getBoolean(columnIndex);
+ case INT32:
+ return input.getInt(columnIndex);
+ case INT64:
+ case TIMESTAMP:
+ return input.getLong(columnIndex);
+ case FLOAT:
+ return input.getFloat(columnIndex);
+ case DOUBLE:
+ return input.getDouble(columnIndex);
+ case TEXT:
+ case STRING:
+ case BLOB:
+ return input.getBinary(columnIndex);
+ case DATE:
+ return input.getLocalDate(columnIndex);
+ default:
+ throw new IllegalArgumentException(String.valueOf(partitionType));
+ }
+ }
+
+ public static void writeValue(ColumnBuilder builder, Object value, Type
type) {
+ if (value == null) {
+ builder.appendNull();
+ return;
+ }
+
+ switch (type) {
+ case BOOLEAN:
+ builder.writeBoolean((Boolean) value);
+ break;
+ case INT32:
+ builder.writeInt((Integer) value);
+ break;
+ case INT64:
+ case TIMESTAMP:
+ builder.writeLong((Long) value);
+ break;
+ case FLOAT:
+ builder.writeFloat((Float) value);
+ break;
+ case DOUBLE:
+ builder.writeDouble((Double) value);
+ break;
+ case TEXT:
+ case STRING:
+ case BLOB:
+ builder.writeBinary((org.apache.tsfile.utils.Binary) value);
+ break;
+ case DATE:
+ builder.writeObject(value);
+ break;
+ default:
+ throw new IllegalArgumentException(String.valueOf(type));
+ }
+ }
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/XCorrTableFunction.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/XCorrTableFunction.java
new file mode 100644
index 00000000000..3f0ce745bae
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/XCorrTableFunction.java
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.udf.builtin.relational.tvf;
+
+import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.i18n.CommonMessages;
+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.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 java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class XCorrTableFunction implements TableFunction {
+
+ public static final String DATA_PARAMETER_NAME = "DATA";
+ public static final String TIMECOL_PARAMETER_NAME = "TIMECOL";
+ private static final String PARTITION_TYPES_PROPERTY = "PARTITION_TYPES";
+
+ @Override
+ public List<ParameterSpecification> getArgumentsSpecifications() {
+ return Arrays.asList(
+
TableParameterSpecification.builder().name(DATA_PARAMETER_NAME).setSemantics().build(),
+ ScalarParameterSpecification.builder()
+ .name(TIMECOL_PARAMETER_NAME)
+ .type(Type.STRING)
+ .build());
+ }
+
+ @Override
+ public TableFunctionAnalysis analyze(Map<String, Argument> arguments) throws
UDFException {
+ // order by column must only be the time column
+ int timeColumnIndex =
+ WindowTVFUtils.checkOrderByColumn(arguments, DATA_PARAMETER_NAME,
TIMECOL_PARAMETER_NAME);
+ TableArgument tableArgument = (TableArgument)
arguments.get(DATA_PARAMETER_NAME);
+
+ List<Integer> partitionIndexes =
WindowTVFUtils.getPartitionIndexes(tableArgument);
+ Set<Integer> excludedIndexes = new HashSet<>(partitionIndexes);
+ excludedIndexes.add(timeColumnIndex);
+
+ List<Type> partitionTypes = new ArrayList<>();
+ DescribedSchema.Builder schemaBuilder = new DescribedSchema.Builder();
+
+ // record the partition columns
+ for (int partitionIndex : partitionIndexes) {
+ Type partitionType = tableArgument.getFieldTypes().get(partitionIndex);
+ partitionTypes.add(partitionType);
+ schemaBuilder.addField(
+ tableArgument.getFieldNames().get(partitionIndex).get(),
partitionType);
+ }
+
+ List<Integer> calculationIndexes =
+ new ArrayList<>(WindowTVFUtils.getCalculationIndexes(tableArgument,
excludedIndexes, null));
+
+ if (calculationIndexes.size() != 2) {
+ throw new SemanticException(
+ String.format(
+ CommonMessages
+
.EXCEPTION_XCORR_REQUIRES_EXACTLY_TWO_CALCULATION_COLUMNS_BUT_FOUND_ARG_2FF8EB0C,
+ calculationIndexes.size()));
+ }
+
+ // XCorr emits one correlation value per lag; the original time column is
used for ordering
+ // only and is not part of the result schema.
+ String firstColumnName =
tableArgument.getFieldNames().get(calculationIndexes.get(0)).get();
+ String secondColumnName =
tableArgument.getFieldNames().get(calculationIndexes.get(1)).get();
+ schemaBuilder.addField(
+ String.format("xcorr(%s, %s)", firstColumnName, secondColumnName),
Type.DOUBLE);
+
+ MapTableFunctionHandle handle =
+ new MapTableFunctionHandle.Builder()
+ .addProperty(PARTITION_TYPES_PROPERTY,
WindowTVFUtils.joinTypes(partitionTypes))
+ .build();
+
+ List<Integer> requiredColumns = new ArrayList<>(partitionIndexes);
+ requiredColumns.add(timeColumnIndex);
+ requiredColumns.addAll(calculationIndexes);
+
+ 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;
+ Type[] partitionTypes =
+ WindowTVFUtils.parseTypes((String)
handle.getProperty(PARTITION_TYPES_PROPERTY));
+
+ return new TableFunctionProcessorProvider() {
+ @Override
+ public TableFunctionDataProcessor getDataProcessor() {
+ return new XCorrDataProcessor(partitionTypes);
+ }
+ };
+ }
+
+ private static class XCorrDataProcessor implements
TableFunctionDataProcessor {
+
+ private static final int INITIAL_CAPACITY = 512;
+ // Cross-correlation is O(n^2) in finish(); keep this far below the filter
+ // function's 65536 so a full partition stays within a few milliseconds.
+ private static final int MAX_COUNT_IN_ONE_PARTITION = 2048;
+
+ private final int partitionColumnCount;
+ private final Type[] partitionTypes;
+ private final Object[] partitionValues;
+
+ private double[] firstValues;
+ private double[] secondValues;
+ private int partitionRowCount;
+
+ private XCorrDataProcessor(Type[] partitionTypes) {
+ this.partitionTypes = partitionTypes;
+ this.partitionColumnCount = partitionTypes.length;
+ this.partitionValues = new Object[partitionColumnCount];
+ this.firstValues = new double[INITIAL_CAPACITY];
+ this.secondValues = new double[INITIAL_CAPACITY];
+ this.partitionRowCount = 0;
+ }
+
+ @Override
+ public void process(
+ Record input,
+ List<ColumnBuilder> properColumnBuilders,
+ ColumnBuilder passThroughIndexBuilder) {
+ if (partitionRowCount >= MAX_COUNT_IN_ONE_PARTITION) {
+ throw new SemanticException(
+ CommonMessages.EXCEPTION_FILTER_FUNCTION_ROW_INDEX_EXCEED_MAXIMUM);
+ }
+
+ if (partitionRowCount == 0) {
+ for (int i = 0; i < partitionColumnCount; i++) {
+ partitionValues[i] =
+ input.isNull(i) ? null : WindowTVFUtils.readValue(input, i,
partitionTypes[i]);
+ }
+ }
+
+ ensureCapacity(partitionRowCount + 1);
+ int firstValueIndex = partitionColumnCount + 1;
+ int secondValueIndex = partitionColumnCount + 2;
+ // Keep the two series aligned. A null value is represented by NaN and
skipped during a pair.
+ firstValues[partitionRowCount] = readFiniteValueOrNaN(input,
firstValueIndex);
+ secondValues[partitionRowCount] = readFiniteValueOrNaN(input,
secondValueIndex);
+ partitionRowCount++;
+ }
+
+ private static double readFiniteValueOrNaN(Record input, int columnIndex) {
+ if (input.isNull(columnIndex)) {
+ return Double.NaN;
+ }
+ double value = input.getDouble(columnIndex);
+ return Double.isFinite(value) ? value : Double.NaN;
+ }
+
+ @Override
+ public void finish(
+ List<ColumnBuilder> properColumnBuilders, ColumnBuilder
passThroughIndexBuilder) {
+ if (partitionRowCount == 0) {
+ return;
+ }
+
+ ColumnBuilder correlationBuilder =
properColumnBuilders.get(partitionColumnCount);
+ // Emit lags in the documented order: -(n - 1), ..., 0, ..., +(n - 1).
+ for (int lag = 1 - partitionRowCount; lag < partitionRowCount; lag++) {
+ int firstStart = Math.max(0, lag);
+ int secondStart = Math.max(0, -lag);
+ int overlapLength = partitionRowCount - Math.abs(lag);
+ double correlation = 0.0;
+ int validPairCount = 0;
+
+ for (int i = 0; i < overlapLength; i++) {
+ double firstValue = firstValues[firstStart + i];
+ double secondValue = secondValues[secondStart + i];
+ if (Double.isFinite(firstValue) && Double.isFinite(secondValue)) {
+ correlation += firstValue * secondValue;
+ validPairCount++;
+ }
+ }
+
+ for (int i = 0; i < partitionColumnCount; i++) {
+ WindowTVFUtils.writeValue(
+ properColumnBuilders.get(i), partitionValues[i],
partitionTypes[i]);
+ }
+ if (validPairCount == 0) {
+ correlationBuilder.appendNull();
+ } else {
+ correlationBuilder.writeDouble(correlation / validPairCount);
+ }
+ }
+ }
+
+ private void ensureCapacity(int requiredCapacity) {
+ if (requiredCapacity <= firstValues.length) {
+ return;
+ }
+ int newCapacity = firstValues.length + (firstValues.length >> 1);
+ while (newCapacity < requiredCapacity) {
+ newCapacity += newCapacity >> 1;
+ }
+ firstValues = Arrays.copyOf(firstValues, newCapacity);
+ secondValues = Arrays.copyOf(secondValues, newCapacity);
+ }
+ }
+}
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
index e0d6b89f1ba..5c96f86e36d 100644
---
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
@@ -21,7 +21,17 @@ package
org.apache.iotdb.commons.udf.builtin.relational.tvf.fft;
import org.apache.iotdb.commons.i18n.QueryMessages;
-/** Computes an in-place 1D forward DFT for interleaved complex double data. */
+/**
+ * Computes in-place 1D FFTs for interleaved complex double data.
+ *
+ * <p>This is an independent implementation and is not a bit-for-bit port of
JTransforms 3.1. For
+ * the same input and frequency-domain operations, its results are
mathematically equivalent to the
+ * original JTransforms-based implementation, but floating-point operation
order may differ.
+ * Therefore, exact bitwise equality is not guaranteed. There is no universal
absolute error bound
+ * independent of input magnitude and transform length; callers should compare
results with a
+ * combined absolute/relative tolerance (1e-9 is a practical baseline for
ordinary finite-valued
+ * LowPass inputs, not a correctness guarantee for every possible input).
+ */
public final class DoubleFFT_1D {
private final int length;
@@ -49,6 +59,28 @@ public final class DoubleFFT_1D {
}
}
+ public void complexInverse(double[] values, boolean scale) {
+ if (values.length < 2 * length) {
+ throw new IllegalArgumentException(
+
QueryMessages.EXCEPTION_INPUT_ARRAY_LENGTH_MUST_BE_AT_LEAST_2_FFT_LENGTH_31DF6A25);
+ }
+
+ // IDFT(x) = conjugate(DFT(conjugate(x))).
+ for (int i = 1; i < 2 * length; i += 2) {
+ values[i] = -values[i];
+ }
+ complexForward(values);
+ for (int i = 1; i < 2 * length; i += 2) {
+ values[i] = -values[i];
+ }
+
+ if (scale) {
+ for (int i = 0; i < 2 * length; i++) {
+ values[i] /= length;
+ }
+ }
+ }
+
private void bluesteinForward(double[] values) {
int convolutionLength = nextPowerOfTwo(2 * length - 1);
double[] a = new double[2 * convolutionLength];
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
index 86f9d6c11fa..feefefc4ce6 100644
---
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
@@ -45,6 +45,45 @@ public class FFT1DTest {
assertArrayEquals(expected, values, 1e-9);
}
+ @Test
+ public void testDoubleComplexInversePowerOfTwoLength() {
+ double[] values = {1.0, 0.5, -2.0, 1.0, 0.0, -1.5, 3.0, 2.0};
+ double[] expected = values.clone();
+
+ DoubleFFT_1D fft = new DoubleFFT_1D(4);
+ fft.complexForward(values);
+ fft.complexInverse(values, true);
+
+ assertArrayEquals(expected, values, 1e-9);
+ }
+
+ @Test
+ public void testDoubleComplexInverseNonPowerOfTwoLength() {
+ double[] values = {1.0, 0.0, 2.0, -0.5, -1.0, 1.5, 0.0, 0.25, 3.0, -2.0};
+ double[] expected = values.clone();
+
+ DoubleFFT_1D fft = new DoubleFFT_1D(5);
+ fft.complexForward(values);
+ fft.complexInverse(values, true);
+
+ assertArrayEquals(expected, values, 1e-9);
+ }
+
+ @Test
+ public void testDoubleComplexInverseWithoutScaling() {
+ double[] values = {1.0, 0.5, -2.0, 1.0, 0.0, -1.5, 3.0, 2.0};
+ double[] expected = values.clone();
+ for (int i = 0; i < expected.length; i++) {
+ expected[i] *= 4;
+ }
+
+ DoubleFFT_1D fft = new DoubleFFT_1D(4);
+ fft.complexForward(values);
+ fft.complexInverse(values, false);
+
+ 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};
