JackieTien97 commented on code in PR #18539:
URL: https://github.com/apache/iotdb/pull/18539#discussion_r3903898142
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/XCorrTableFunction.java:
##########
@@ -0,0 +1,229 @@
+package org.apache.iotdb.commons.udf.builtin.relational.tvf;
Review Comment:
[P1] Add the ASF license header before merging. Running `mvn
apache-rat:check -pl iotdb-core/node-commons -DskipTests` reports this file as
the sole unapproved file, and a normal `clean verify` stops in `node-commons`
before the integration tests can run.
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/XCorrTableFunction.java:
##########
@@ -0,0 +1,229 @@
+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;
+
+import static
org.apache.iotdb.commons.udf.builtin.relational.tvf.FilterTransferTableFunction.FilterTransferDataProcessor.MAX_COUNT_IN_ONE_PARTITION;
+
+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;
+
+ 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++) {
Review Comment:
[P1] The accepted partition limit makes this loop impractical. Summing
`overlapLength` over all lags yields exactly `n^2` pair iterations; at the
shared limit of 65,536 that is 4,294,967,296 iterations inside one synchronous
`finish()` call. Please use an FFT-based correlation (and a mask correlation
for valid-pair counts), or introduce a much smaller XCorr-specific bound plus
cancellation/yield checks.
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/XCorrTableFunction.java:
##########
@@ -0,0 +1,229 @@
+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;
+
+import static
org.apache.iotdb.commons.udf.builtin.relational.tvf.FilterTransferTableFunction.FilterTransferDataProcessor.MAX_COUNT_IN_ONE_PARTITION;
+
+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);
Review Comment:
[P1] Please include the lag as a proper result column (for example, `lag
INT64`). `finish()` emits one row for each lag from `-(n - 1)` through `n - 1`,
but the schema only exposes the partition keys and the correlation value. Rows
within a partition are therefore indistinguishable with respect to lag, and SQL
does not guarantee the producer's physical row order. The new tests exhibit
this too: they can only order by partition columns, leaving all lag rows tied.
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/WindowTVFUtils.java:
##########
@@ -58,4 +89,152 @@ public static int findColumnIndex(
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).get();
Review Comment:
[P2] Field names are optional for table arguments. An unaliased numeric
expression such as `SELECT time, s1 + s2 FROM t` reaches this code with
`Optional.empty()`, so this `.get()` leaks `NoSuchElementException: No value
present` instead of a SQL semantic error. Please use `orElseThrow` with a
localized `SemanticException` (as FFT does), or define a generated output name.
##########
integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java:
##########
@@ -1347,4 +1347,236 @@ 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.66666666666669,101.00000000000001,",
Review Comment:
[P2] These FFT results should not be asserted through exact JDBC string
equality. `tableResultSetEqualTest` compares the full string, while
`DoubleFFT_1D` explicitly states that floating-point operation order may differ
and recommends absolute/relative tolerance. Values such as this one can vary by
a few ulps across architectures or JDKs. Please compare numeric columns with an
appropriate combined tolerance.
##########
integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBWindowTVFIT.java:
##########
@@ -1347,4 +1347,236 @@ 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.66666666666669,101.00000000000001,",
+ "AAPL,2021-01-01T09:09:00.000Z,101.66666666666667,101.0,",
+ "TESL,2021-01-01T09:06:00.000Z,199.0,211.99999999999991,",
+ "TESL,2021-01-01T09:07:00.000Z,199.0,212.0,",
+
"TESL,2021-01-01T09:15:00.000Z,199.00000000000003,211.99999999999991,"
+ };
+ tableResultSetEqualTest(
+ "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,"
+ };
+ tableResultSetEqualTest(
+ "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,9.999999999999998,",
+ "F1,device_1,1970-01-01T00:00:00.005Z,10.000000000000002,",
+ "F1,device_1,1970-01-01T00:00:00.009Z,9.999999999999998,",
+ "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,"
+ };
+ tableResultSetEqualTest(
+ "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.666666666666661,1.7936858794783102E-15,",
+
"AAPL,2021-01-01T09:07:00.000Z,1.3333333333333224,-1.3450096248017031E-14,",
+
"AAPL,2021-01-01T09:09:00.000Z,0.33333333333333814,1.1656410368538743E-14,",
+
"TESL,2021-01-01T09:06:00.000Z,1.0000000000000255,-109.99999999999996,",
+
"TESL,2021-01-01T09:07:00.000Z,2.999999999999977,-10.000000000000027,",
+ "TESL,2021-01-01T09:15:00.000Z,-4.0000000000000036,120.0,"
+ };
+ tableResultSetEqualTest(
+ "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,"
+ };
+ tableResultSetEqualTest(
+ "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,-7.401486830834377E-16,",
+ "F1,device_1,1970-01-01T00:00:00.005Z,4.999999999999998,",
+ "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,"
+ };
+ tableResultSetEqualTest(
+ "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 => table1 PARTITION BY device_id ORDER BY
time, TIMECOL => 'time')",
Review Comment:
[P1] This test does not reach the calculation-column-count validation it is
intended to cover. `table1` also contains `s3 STRING`, and
`getCalculationIndexes` validates every candidate type before XCorr checks the
list size, so the actual error is the unsupported-STRING error. I reproduced
this in both TableSimpleIT and TableClusterIT (39 tests, 1 failure in each).
Please build the input from three numeric expressions/columns, or change the
expected behavior consistently.
##########
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);
Review Comment:
[P2] This user-visible output-column template must go through the module's
i18n messages instead of being an inline `String.format` literal, per the
repository's i18n rule. The same issue exists in
`HighPassTableFunction.convertColumnName` and the XCorr output-name template.
Please add matching en/zh constants using the required deterministic hash
suffixes and reference them here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
