From e11123d500f9a22fec8aaa7571941b709a87da92 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Tue, 25 Aug 2026 17:26:22 +0800 Subject: [PATCH 1/3] finish the LowPassTableFunction --- .../analyzer/StatementAnalyzer.java | 1 + .../relational/planner/RelationPlanner.java | 13 +- .../iotdb/commons/i18n/CommonMessages.java | 6 +- .../iotdb/commons/i18n/CommonMessages.java | 5 +- .../function/TableBuiltinTableFunction.java | 6 +- .../tvf/FilterTransferTableFunction.java | 317 ++++++++++++++++++ .../relational/tvf/LowPassTableFunction.java | 77 +++++ .../relational/tvf/M4TableFunction.java | 87 +---- .../relational/tvf/WindowTVFUtils.java | 144 ++++++++ .../relational/tvf/fft/DoubleFFT_1D.java | 34 +- .../builtin/relational/tvf/fft/FFT1DTest.java | 39 +++ 11 files changed, 644 insertions(+), 85 deletions(-) create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FilterTransferTableFunction.java create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/LowPassTableFunction.java 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 9a54519953d6a..4310e747069be 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,7 @@ private ArgumentsAnalysis analyzeArguments( private boolean isPartitionColumnsProvidedByProperSchema(String functionName) { return TableBuiltinTableFunction.M4.getFunctionName().equalsIgnoreCase(functionName) + || TableBuiltinTableFunction.LOWPASS.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 7fc367b378592..9a4de1ca2ddf9 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 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,12 @@ 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.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 94194a086b47c..0f85486013dd1 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 @@ -222,7 +222,11 @@ private CommonMessages() {} 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"; } 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 07d886a44f3d4..c3584f8e3bb06 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,8 @@ private 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 = "分区行数超过了最大限制"; } 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 decaf8dc4669e..11b995cc32ee7 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,6 +27,7 @@ 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.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; @@ -49,7 +50,8 @@ public enum TableBuiltinTableFunction { FFT("fft"), FORECAST("forecast"), PATTERN_MATCH("pattern_match"), - CLASSIFY("classify"); + CLASSIFY("classify"), + LOWPASS("lowpass"); private final String functionName; @@ -99,6 +101,8 @@ public static TableFunction getBuiltinTableFunction(String functionName) { return new ForecastTableFunction(); case "classify": return new ClassifyTableFunction(); + case "lowpass": + return new LowPassTableFunction(); 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 0000000000000..4336feb1338ae --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/FilterTransferTableFunction.java @@ -0,0 +1,317 @@ +/* + * 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"; + protected static final Set ALLOWED_TYPES = + Set.of(Type.DOUBLE, Type.FLOAT, Type.INT32, Type.INT64); + + @Override + public List 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 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 partitionIndexes = WindowTVFUtils.getPartitionIndexes(tableArgument); + Set excludedIndexes = new HashSet<>(partitionIndexes); + excludedIndexes.add(timeColumnIndex); + + List partitionTypes = new ArrayList<>(); + List 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 + 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(); + if (!ALLOWED_TYPES.contains(type)) { + throw new SemanticException( + String.format(CommonMessages.EXCEPTION_NOT_ALLOWED_COLUMNS, columnName, type)); + } + + // all the result column would be the double type + calculationIndexes.add(i); + 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 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); + + /** every partition data (may include multiple columns) responding to a DataProcessor */ + protected abstract static class FilterTransferDataProcessor + implements TableFunctionDataProcessor { + + protected static final int INITIAL_CAPACITY = 512; + private static final int MAX_COUNT_IN_ONE_PARTITION = 65536; + + private final double wpass; + private int partitionRowIndex; + + 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.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 properColumnBuilders, + ColumnBuilder passThroughIndexBuilder) { + if (partitionRowIndex >= MAX_COUNT_IN_ONE_PARTITION) { + throw new SemanticException( + CommonMessages.EXCEPTION_FILTER_FUNCTION_ROW_INDEX_EXCEED_MAXIMUM); + } + if (partitionRowIndex == 0) { + capturePartitionValues(input); + } + collectTimeColumnValue(input); + collectCalculationValues(input, partitionRowIndex); + partitionRowIndex++; + } + + 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 (partitionRowIndex >= partitionTimestamps.length) { + int newCapacity = partitionTimestamps.length + (partitionTimestamps.length >> 2); + partitionTimestamps = Arrays.copyOf(partitionTimestamps, newCapacity); + } + partitionTimestamps[partitionRowIndex] = input.getLong(timeColumnIndex); + } + + private void collectCalculationValues(Record input, int partitionRowIndex) { + for (int i = 0; i < calculationColumnContainers.length; i++) { + 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 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 < partitionRowIndex; rowIndex++) { + WindowTVFUtils.writeValue(partitionColumnBuilder, partitionValue, partitionType); + } + } + + // collect the time column + ColumnBuilder timeColumnBuilder = properColumnBuilders.get(timeColumnIndex); + for (int rowIndex = 0; rowIndex < partitionRowIndex; 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 < partitionRowIndex; rowIndex++) { + properColumnBuilder.appendNull(); + } + return; + } + double[] temp = filterTransform(columnContainer, size, wpass); + // collect the value after the transformation + int validValueIndex = 0; + for (int i = 0; i < partitionRowIndex; 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/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 0000000000000..b563af717a435 --- /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 733696b62123d..7b508fcb53d65 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.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 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 getArgumentsSpecifications() { @@ -113,31 +99,20 @@ public List getArgumentsSpecifications() { @Override public TableFunctionAnalysis analyze(Map 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 partitionIndexes = getPartitionIndexes(tableArgument); + WindowTVFUtils.checkOrderByColumn(arguments, DATA_PARAMETER_NAME, TIMECOL_PARAMETER_NAME); + TableArgument tableArgument = (TableArgument) arguments.get(DATA_PARAMETER_NAME); + List partitionIndexes = WindowTVFUtils.getPartitionIndexes(tableArgument); Set 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 participantIndexes = new ArrayList<>(); List partitionTypes = new ArrayList<>(); List 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 TableFunctionAnalysis analyze(Map arguments) throws UDF .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 TableFunctionProcessorProvider getProcessorProvider( 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 TableFunctionDataProcessor getDataProcessor() { }; } - 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 getPartitionIndexes(TableArgument tableArgument) { - List 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 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 3271a30239abc..5b8fdbb185ce2 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,44 @@ 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; public class WindowTVFUtils { + + private static final Set SUPPORTED_PARTITION_TYPES = + new HashSet<>( + Arrays.asList( + Type.BOOLEAN, + Type.INT32, + Type.INT64, + Type.FLOAT, + Type.DOUBLE, + Type.TEXT, + Type.TIMESTAMP, + Type.DATE, + Type.BLOB, + Type.STRING)); + /** * Find the index of the column in the table argument. * @@ -58,4 +85,121 @@ 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 getPartitionIndexes(TableArgument tableArgument) { + List indexes = new ArrayList<>(); + for (String partitionColumn : tableArgument.getPartitionBy()) { + indexes.add(findColumnIndex(tableArgument, partitionColumn, SUPPORTED_PARTITION_TYPES)); + } + return indexes; + } + + public static String joinTypes(List types) { + StringBuilder builder = new StringBuilder(); + for (int i = 0; i < types.size(); i++) { + if (i > 0) { + builder.append(','); + } + builder.append(types.get(i).name()); + } + return builder.toString(); + } + + /** check the order by column is the timeColumn */ + public static int checkOrderByColumn( + Map 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/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 e0d6b89f1ba7b..5c96f86e36dc8 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 @@ 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. + * + *

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 void complexForward(double[] values) { } } + 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 86f9d6c11fa96..feefefc4ce643 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 void testDoubleComplexForwardNonPowerOfTwoLength() { 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}; From 07f2e44a97bc493ed008317540097ff06eda2536 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Fri, 28 Aug 2026 11:50:40 +0800 Subject: [PATCH 2/3] finish the HighPassTableFunction --- .../analyzer/StatementAnalyzer.java | 1 + .../relational/planner/RelationPlanner.java | 1 + .../iotdb/commons/i18n/CommonMessages.java | 249 ++++++++++++------ .../function/TableBuiltinTableFunction.java | 6 +- .../tvf/FilterTransferTableFunction.java | 13 +- .../relational/tvf/HighPassTableFunction.java | 81 ++++++ 6 files changed, 271 insertions(+), 80 deletions(-) create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/HighPassTableFunction.java 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 4310e747069be..ac989291d9b0d 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 @@ -5924,6 +5924,7 @@ 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.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 9a4de1ca2ddf9..420084fdd3d6a 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 @@ -1611,6 +1611,7 @@ public RelationPlan visitTableFunctionInvocation(TableFunctionInvocation node, V private boolean needAddPartitionColumn(String functionName) { return !TableBuiltinTableFunction.M4.getFunctionName().equalsIgnoreCase(functionName) && !TableBuiltinTableFunction.LOWPASS.getFunctionName().equalsIgnoreCase(functionName) + && !TableBuiltinTableFunction.HIGHPASS.getFunctionName().equalsIgnoreCase(functionName) && !TableBuiltinTableFunction.FFT.getFunctionName().equalsIgnoreCase(functionName); } 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 0f85486013dd1..5746db9a1d500 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,95 +135,191 @@ private 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_FILTER_FUNCTION_WPASS_VALIDATION = "the value of wpass should be in (0, 1)"; + 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 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 = 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 11b995cc32ee7..a4710c1e2bc3f 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,6 +27,7 @@ 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; @@ -51,7 +52,8 @@ public enum TableBuiltinTableFunction { FORECAST("forecast"), PATTERN_MATCH("pattern_match"), CLASSIFY("classify"), - LOWPASS("lowpass"); + LOWPASS("lowpass"), + HIGHPASS("highpass"); private final String functionName; @@ -103,6 +105,8 @@ public static TableFunction getBuiltinTableFunction(String functionName) { return new ClassifyTableFunction(); case "lowpass": return new LowPassTableFunction(); + case "highpass": + return new HighPassTableFunction(); 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 index 4336feb1338ae..724152006369b 100644 --- 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 @@ -159,7 +159,15 @@ public abstract TableFunctionProcessorProvider getProcessorProvider( protected abstract String convertColumnName(String columnName); - /** every partition data (may include multiple columns) responding to a DataProcessor */ + /** + * Processes one complete partition (which may contain multiple calculation columns). + * + *

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 { @@ -228,6 +236,7 @@ private void collectTimeColumnValue(Record input) { 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)) { @@ -279,7 +288,7 @@ private void transformSingleColumn( return; } double[] temp = filterTransform(columnContainer, size, wpass); - // collect the value after the transformation + // Restore the transformed values to their original rows; excluded rows stay null. int validValueIndex = 0; for (int i = 0; i < partitionRowIndex; i++) { if (columnContainer.validRows.get(i)) { 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 0000000000000..88fa37bee62b7 --- /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; + } + } +} From 154a9020626013cf71e6a56918dc1758413a6797 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Fri, 28 Aug 2026 15:10:09 +0800 Subject: [PATCH 3/3] finish the XCorrTableFunction --- .../analyzer/StatementAnalyzer.java | 1 + .../relational/planner/RelationPlanner.java | 1 + .../iotdb/commons/i18n/CommonMessages.java | 3 + .../iotdb/commons/i18n/CommonMessages.java | 3 + .../function/TableBuiltinTableFunction.java | 6 +- .../tvf/FilterTransferTableFunction.java | 47 ++-- .../relational/tvf/WindowTVFUtils.java | 35 +++ .../relational/tvf/XCorrTableFunction.java | 229 ++++++++++++++++++ 8 files changed, 295 insertions(+), 30 deletions(-) create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/udf/builtin/relational/tvf/XCorrTableFunction.java 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 ac989291d9b0d..b5c394539f217 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 @@ -5925,6 +5925,7 @@ 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 420084fdd3d6a..6f5f3c278ed3d 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 @@ -1612,6 +1612,7 @@ 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); } 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 5746db9a1d500..870d88ae19693 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 @@ -324,4 +324,7 @@ private CommonMessages() {} "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."; } 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 c3584f8e3bb06..4964b77f957af 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 @@ -221,4 +221,7 @@ private CommonMessages() {} 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 列。"; } 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 a4710c1e2bc3f..818747bb71494 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 @@ -33,6 +33,7 @@ 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; @@ -53,7 +54,8 @@ public enum TableBuiltinTableFunction { PATTERN_MATCH("pattern_match"), CLASSIFY("classify"), LOWPASS("lowpass"), - HIGHPASS("highpass"); + HIGHPASS("highpass"), + XCORR("xcorr"); private final String functionName; @@ -107,6 +109,8 @@ public static TableFunction getBuiltinTableFunction(String functionName) { 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 index 724152006369b..0fe09c60cb7f9 100644 --- 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 @@ -58,8 +58,6 @@ public abstract class FilterTransferTableFunction implements TableFunction { protected static final String PARTITION_TYPES_PROPERTY = "PARTITION_TYPES"; protected static final String CALCULATION_COLUMN_COUNT_PROPERTY = "CALCULATION_COLUMN_COUNT"; - protected static final Set ALLOWED_TYPES = - Set.of(Type.DOUBLE, Type.FLOAT, Type.INT32, Type.INT64); @Override public List getArgumentsSpecifications() { @@ -111,21 +109,11 @@ public TableFunctionAnalysis analyze(Map arguments) throws UDF // record the time column schemaBuilder.addField(tableArgument.getFieldNames().get(timeColumnIndex), Type.TIMESTAMP); // record the calculation columns, only double, float, int32, and int64 are allowed - 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(); - if (!ALLOWED_TYPES.contains(type)) { - throw new SemanticException( - String.format(CommonMessages.EXCEPTION_NOT_ALLOWED_COLUMNS, columnName, type)); - } - - // all the result column would be the double type - calculationIndexes.add(i); - schemaBuilder.addField(convertColumnName(columnName), Type.DOUBLE); - } + calculationIndexes.addAll( + WindowTVFUtils.getCalculationIndexes( + tableArgument, + excludedIndexes, + columnName -> schemaBuilder.addField(convertColumnName(columnName), Type.DOUBLE))); if (calculationIndexes.isEmpty()) { throw new SemanticException(CommonMessages.EXCEPTION_NO_CALCULATE_COLUMNS); @@ -172,10 +160,10 @@ protected abstract static class FilterTransferDataProcessor implements TableFunctionDataProcessor { protected static final int INITIAL_CAPACITY = 512; - private static final int MAX_COUNT_IN_ONE_PARTITION = 65536; + protected static final int MAX_COUNT_IN_ONE_PARTITION = 65536; private final double wpass; - private int partitionRowIndex; + private int partitionRowCount; private final int partitionColumnCount; private final int timeColumnIndex; @@ -192,6 +180,7 @@ protected FilterTransferDataProcessor( 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]; @@ -207,16 +196,16 @@ public void process( Record input, List properColumnBuilders, ColumnBuilder passThroughIndexBuilder) { - if (partitionRowIndex >= MAX_COUNT_IN_ONE_PARTITION) { + if (partitionRowCount >= MAX_COUNT_IN_ONE_PARTITION) { throw new SemanticException( CommonMessages.EXCEPTION_FILTER_FUNCTION_ROW_INDEX_EXCEED_MAXIMUM); } - if (partitionRowIndex == 0) { + if (partitionRowCount == 0) { capturePartitionValues(input); } collectTimeColumnValue(input); - collectCalculationValues(input, partitionRowIndex); - partitionRowIndex++; + collectCalculationValues(input, partitionRowCount); + partitionRowCount++; } private void capturePartitionValues(Record input) { @@ -227,11 +216,11 @@ private void capturePartitionValues(Record input) { } private void collectTimeColumnValue(Record input) { - if (partitionRowIndex >= partitionTimestamps.length) { + if (partitionRowCount >= partitionTimestamps.length) { int newCapacity = partitionTimestamps.length + (partitionTimestamps.length >> 2); partitionTimestamps = Arrays.copyOf(partitionTimestamps, newCapacity); } - partitionTimestamps[partitionRowIndex] = input.getLong(timeColumnIndex); + partitionTimestamps[partitionRowCount] = input.getLong(timeColumnIndex); } private void collectCalculationValues(Record input, int partitionRowIndex) { @@ -256,14 +245,14 @@ public void finish( ColumnBuilder partitionColumnBuilder = properColumnBuilders.get(columnIndex); Object partitionValue = partitionValues[columnIndex]; Type partitionType = partitionTypes[columnIndex]; - for (int rowIndex = 0; rowIndex < partitionRowIndex; rowIndex++) { + 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 < partitionRowIndex; rowIndex++) { + for (int rowIndex = 0; rowIndex < partitionRowCount; rowIndex++) { timeColumnBuilder.writeLong(partitionTimestamps[rowIndex]); } @@ -282,7 +271,7 @@ private void transformSingleColumn( ColumnBuilder properColumnBuilder) { int size = columnContainer.validValueCount; if (size == 0) { - for (int rowIndex = 0; rowIndex < partitionRowIndex; rowIndex++) { + for (int rowIndex = 0; rowIndex < partitionRowCount; rowIndex++) { properColumnBuilder.appendNull(); } return; @@ -290,7 +279,7 @@ private void transformSingleColumn( 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 < partitionRowIndex; i++) { + for (int i = 0; i < partitionRowCount; i++) { if (columnContainer.validRows.get(i)) { properColumnBuilder.writeDouble(temp[2 * validValueIndex]); validValueIndex++; 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 5b8fdbb185ce2..605f109828de2 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 @@ -40,9 +40,13 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.function.Consumer; public class WindowTVFUtils { + private static final Set ALLOWED_CALCULATION_TYPES = + Set.of(Type.DOUBLE, Type.FLOAT, Type.INT32, Type.INT64); + private static final Set SUPPORTED_PARTITION_TYPES = new HashSet<>( Arrays.asList( @@ -103,6 +107,37 @@ public static List getPartitionIndexes(TableArgument tableArgument) { return indexes; } + /** + * Collect calculation-column indexes after excluding partition and time columns. + * + *

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 getCalculationIndexes( + TableArgument tableArgument, + Set excludedIndexes, + Consumer calculationColumnConsumer) { + List 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(); + 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 types) { StringBuilder builder = new StringBuilder(); for (int i = 0; i < types.size(); i++) { 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 0000000000000..d3107a732d335 --- /dev/null +++ b/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 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 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 partitionIndexes = WindowTVFUtils.getPartitionIndexes(tableArgument); + Set excludedIndexes = new HashSet<>(partitionIndexes); + excludedIndexes.add(timeColumnIndex); + + List 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 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 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 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 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); + } + } +}