Implementation of LowPass, HighPass, and XCorr Table Functions

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

@@ -31,17 +31,23 @@
 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 void testM4RejectsNonTimestampTimecol() {
         "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());
+    }
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java

@@ -5923,6 +5923,9 @@ private ArgumentsAnalysis analyzeArguments(
 
     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);
     }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java

@@ -1572,12 +1572,7 @@ public RelationPlan visitTableFunctionInvocation(TableFunctionInvocation node, V
                 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 RelationPlan visitTableFunctionInvocation(TableFunctionInvocation node, V
     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()